Cloudera Operational Database（COD）は、Cloudera Data Platform（CDP）のエクスペリエンスとして利用できるマネージドdbPaaSソリューションです。 Apache HBase APIを使用したNoSQLKey-ValueおよびJDBCを使用したリレーショナルSQL（Apache Phoenix経由）を使用したマルチモーダルクライアントアクセスを提供します。後者は、MySQL、Postgresなどを使用するアプリケーションの構築に慣れている開発者がCODにアクセスできるようにします。CODの主な利点は次のとおりです。
- 自動スケーリング–クラスターのワークロード使用率に基づいており、まもなくクラスターをスケールアップ/スケールダウンできるようになります
- 自動調整–既存のインフラストラクチャフットプリント内でのパフォーマンスの向上。
- 自動修復–運用上の問題を自動的に解決します（近日公開予定）。
このブログでは、CODをバックエンドシステムとして簡単に使用して、単純なWebアプリケーションのデータと画像を保存する方法を紹介します。このアプリケーションを構築するために、CODの基盤となるコンポーネントの1つであるPhoenixをFlaskとともに使用します。画像の保存には、MOB（ミディアムオブジェクト）と呼ばれるHBase（Apache Phoenixバックエンドストレージ）機能を使用します。 MOBを使用すると、100k〜10MBの値をすばやく読み書きできます。
*開発の使いやすさのために、CODの代わりにPhoenixクエリサーバーを使用することもできます。クエリサーバーはフェニックスの小さなビルドであり、開発のみを目的としており、ビルドごとにデータが削除されます。
すべてのコードは私のgithubリポジトリにあります。
手順：
1. Cloudera Management Consoleにログインし、オペレーショナルデータベースエクスペリエンスを選択します
2.環境を選択し、DBに名前を付けます
3. DBが起動したら、シンJDBCクライアントからURLを取得します
4.CDPワークロードパスワードを設定します
5.プロジェクトのgitリポジトリとインストール要件のクローンを作成します： $ pip install -r Requirements.txt
6. appフォルダーに移動し、「setup.py」を実行します。これにより、ユーザーとその画像の3つのレコードを含むテーブルが作成されます $ python setup.py
7.Webアプリケーションを起動するためにフラスコWebサーバーを実行します。$ FLASK_APP =app.py python -mフラスコ実行–port =8888 –host =127.0.0.1 –reload –with-threads –debugger
8.ブラウザでhttp：// localhost：8888/usersにアクセスします。アプリケーションが実行されているのを確認できるはずです。そのように簡単です。
コードを確認する
1. Schemaクラスは、基本的に接続の詳細を保持し、テーブルメソッドを作成および削除します。ご覧のとおり、「写真」列はVARBINARYタイプであり、HBaseのMOBオブジェクトに変換されます。
import phoenixdb
import phoenixdb.cursor
class Schema:
def __init__(self):
opts = {}
opts['authentication'] = 'BASIC'
opts['avatica_user'] = '<cod workload username>'
opts['avatica_password'] = '<cod workload pw>'
database_url = "<cod thin jdbc url>"
self.TABLENAME = "users"
self.conn = phoenixdb.connect(database_url, autocommit=True,**opts)
self.curs = self.conn.cursor()
def create_users_table(self):
query = """
CREATE TABLE IF NOT EXISTS """+self.TABLENAME+""" (
username VARCHAR NOT NULL,
firstname VARCHAR,
lastname VARCHAR,
telephone VARCHAR,
message VARCHAR,
email VARCHAR,
photo VARBINARY,
photo_name VARCHAR,
photo_type VARCHAR,
photo_chars VARCHAR
CONSTRAINT my_pk PRIMARY KEY (username))
"""
self.curs.execute(query)
def drop_users_table(self):
query = "DROP TABLE "+self.TABLENAME
self.curs.execute(query) 2ユーザークラスは、Phoenixでのすべてのアプリケーション操作を担当します。画像トランザクションの更新/挿入（フェニックス言語でのアップサート）、削除、一覧表示、および処理を行うことができます：
import phoenixdb
from schema import Schema
import json
class UsersModel:
TABLENAME = "users"
def __init__(self):
db = Schema()
self.conn=db.conn
self.curs=db.curs
def upsert(self, params):
sql = "upsert into " + self.TABLENAME + \
" (username ,message,telephone,firstname,lastname,email) \
values (?,?,?,?,?,?)"
data = (params.get('username'),params.get('message'),\
params.get('telephone'),params.get('firstname'),\
params.get('lastname'),params.get('email'))
results = self.curs.execute(sql,data)
return results
def upsert_photo(self, params):
if params.get('photo') is None:
photo = bytes('','utf-8')
else:
photo = params.get('photo')
sql = "upsert into " + self.TABLENAME + \
" (username, photo,photo_name) values (?,?,?)"
data = (params.get('username'),photo, params.get('photo_name'))
results = self.curs.execute(sql,data)
return results
def delete(self, username):
query = f"DELETE from {self.TABLENAME} " \
f"WHERE username = {username}"
self.curs.execute(query)
def list_items(self, where_clause="",format="json"):
query = f"SELECT username ,email,message,telephone,firstname,\
lastname,photo_name " \
f"from {self.TABLENAME} WHERE " + where_clause
self.curs.execute(query)
if format=="json":
r = [dict((self.curs.description[i][0].lower(), value) \
for i, value in enumerate(row)) for row in \
self.curs.fetchall()]
self.conn.close()
data={'data': r }
return json.dumps(data)
result_set=self.curs.fetchall()
result = [{column: row[i]
for i, column in enumerate(result_set[0].keys())}
for row in result_set]
return result
def get_image(self, username):
query = f"SELECT photo,photo_name " \
f"from {self.TABLENAME} WHERE username='"+username+"'"
self.curs.execute(query)
row = self.curs.fetchone()
return row 3. app.pyは、アプリケーションのメインルーターです。これには、ユーザー入力によるすべての処理と、それらを接続メソッドにルーティングすることが含まれています。使いやすさのために画像の処理を分離しました。そうすることで、ユーザー向けに特定の画像を取得できます。
from flask import Flask, request, send_file ,jsonify,render_template
import phoenixdb
import io
from users import UsersModel
from schema import Schema
import json
app = Flask(__name__)
@app.after_request
def add_headers(response):
response.headers['Access-Control-Allow-Origin'] = '*'
response.headers['Access-Control-Allow-Headers'] = \
"Content-Type, Access-Control-Allow-Headers, Authorization, \
X-Requested-With"
response.headers['Access-Control-Allow-Methods']= "POST, GET, PUT, \
DELETE, OPTIONS"
response.headers['Allow']= "POST, GET, PUT, OPTIONS"
return response
@app.route("/")
def hello():
return "Hello World!"
@app.route("/users")
def return_form():
return render_template("users.html")
@app.route("/handle_data",methods=['POST'])
def handle_data():
if request.method == 'POST':
username = request.form['username']
firstname = request.form['firstname']
lastname = request.form['lastname']
email = request.form['email']
telephone = request.form['telephone']
message = request.form['message']
photo = request.files['photo']
photo_bytes = photo.read()
model=Schema()
usersmodel=UsersModel()
data = {'username':f"{username}",'firstname':f"{firstname}",\
'lastname':f"{lastname}",'telephone':f"{telephone}",\
'message':f"{message}"}
photo_data = {'username':f"{username}",\
'photo':photo_bytes,\
'photo_name':f"{photo.filename}"}
usersmodel.upsert(data)
usersmodel.upsert_photo(photo_data)
return render_template('users.html')
else:
return render_template('users.html')
@app.route("/get_users",methods=['GET'])
def get_users():
if request.method == 'GET':
usersmodel=UsersModel()
users = usersmodel.list_items("1=1")
return users
@app.route("/get_image",methods=['GET'])
def get_image():
if request.method == 'GET':
username = request.args.get('username')
usersmodel=UsersModel()
imagedb = usersmodel.get_image(username)
return send_file(io.BytesIO(imagedb[0]),mimetype='image/png', \
attachment_filename=imagedb[1])
if __name__ == "__main__":
Schema()
app.run(debug=True, port=8888) 次のステップでは、このgithubリポジトリを使用してアプリケーションをテストできます。
お役に立てば幸いです。ハッピーコーディング!!