GCP Bigtable の操作 - saito-shion technology diary

GCP の Bigtable を操作していく。

前回、Bigtable インスタンスを作成することができた。

続いて、Bigtable インスタンスにデータを永続化できるか確認しようと思う。ここでは Python のプログラムを作成して、Bigtable の動作確認を行う。

前回はこちら→GCPを始める～Cloud Bigtable～

前提条件

VPC ネットワークを事前に作成していること。
ファイアウォールルールを事前に作成していること。
VM インスタンスを事前に作成していること。
※[ネットワークインタフェース] には、app-networkを設定してください。
Bigtable インスタンスを事前に作成していること。

事前準備（アクセススコープ）

[VM インスタンス] ページから、VM インスタンスを選択し、[編集] をクリックします。
[アクセススコープ] を [すべての Cloud API に完全アクセス権を許可] に設定します。
※[各 API にアクセス権を設定] で、Bigtable のみ個別に設定でもOK。

事前準備（cbt）

[VM インスタンス] ページから、VM へ SSH します。
下記のコマンドを実行します。
※エラーとなった場合は、コンソールに表示されるコマンドを実行してください。

$ sudo gcloud components update
$ sudo gcloud components install cbt

エラーとなったので実際に実行したコマンドも記載する。

$ sudo yum makecache && sudo yum update kubectl google-cloud-sdk-minikube google-cloud-sdk google-cloud-sdk-app-engine-grpc google-cloud-sdk-kind google-cloud-sdk-pubsub-emulator google-cloud-sdk-app-engine-go google-cloud-sdk-firestore-emulator google-cloud-sdk-cloud-build-local google-cloud-sdk-datastore-emulator google-cloud-sdk-kpt google-cloud-sdk-app-engine-python google-cloud-sdk-skaffold google-cloud-sdk-cbt google-cloud-sdk-bigtable-emulator google-cloud-sdk-app-engine-python-extras google-cloud-sdk-datalab google-cloud-sdk-app-engine-java
$ sudo yum install google-cloud-sdk-cbt

事前準備（Bigtable インスタンスへの接続）

自分のプロジェクトとインスタンスを使用するように cbt を構成します。そのためには、.cbtrcファイルを作成します。
※project-id は Bigtable インスタンスの作成先であるプロジェクト ID に置き換えます。
※instance-id は Bigtable のインスタンス ID に置き換えます。

$ echo project = project-id > ~/.cbtrc
$ echo instance = instance-id >> ~/.cbtrc

f:id:saito_shion:20200318154604p:plain

テーブル作成

テーブルをhitsという名前で作成し、テーブルの一覧を表示します。

$ cbt createtable hits
$ cbt ls

f:id:saito_shion:20200319155945p:plain 1 つの列ファミリーをhits_family_1という名前で追加し、列ファミリーを一覧表示します。

$ cbt createfamily hits hits_family_1
$ cbt ls hits

f:id:saito_shion:20200323144246p:plain

Python アプリケーションの準備

下記のコマンドを実行してインストールを行う。

$ sudo yum install python3
$ sudo pip3 install flask google-cloud-bigtable google-cloud-core

アプリケーションファイルを作成する。

$ touch app.py

下記をapp.pyにコピーアンドペーストする。
※project_id は、GCP のプロジェクト ID を指定してください。
※instance_id は、Bigtable のインスタンス ID を指定してください。

import datetime
from flask import Flask

from google.cloud import bigtable
from google.cloud.bigtable import column_family
from google.cloud.bigtable import row_filters

app = Flask(__name__)
client = bigtable.Client(project='project_id', admin=True)
instance = client.instance('instance_id')

def get_hit_count():
    result = '0'
    table = instance.table('hits')
    column_family_id = 'hits_family_1'
    row_key = 'hits_value'.encode()

    if not table.exists():
        result = '-1'
    else:
        # table read
        row_filter = row_filters.CellsColumnLimitFilter(1)
        row = table.read_row(row_key, row_filter)

        count = '0'
        if row:
            cell_values = row.cell_values(column_family_id, row_key)
            for value, timestamp in cell_values:
                count = value.decode('utf-8')

        new_value = int(count) + 1

        # table write
        rows = []
        row = table.row(row_key)
        row.set_cell(column_family_id, row_key, str(new_value), timestamp=datetime.datetime.utcnow())
        rows.append(row)
        table.mutate_rows(rows)

        result = str(new_value)
    
    return result

@app.route('/')
def hello():
    count = get_hit_count()
    return 'Hello World! I have been seen {0} times.'.format(int(count))

環境変数を設定する。

$ export FLASK_APP=app.py
$ export FLASK_RUN_HOST=0.0.0.0

FLASK を実行する。

$ flask run

動作確認

Webブラウザでhttp://your ip address:5000へアクセスする。 f:id:saito_shion:20200323161139p:plain

コード解説

基本情報

Bigtable へのアクセスオブジェクトを生成している（アドミン権限で）。

client = bigtable.Client(project='project_id', admin=True)

Bigtable のインスタンスへアクセスするオブジェクトを生成している。

instance = client.instance('instance_id')

hits テーブルへアクセスすることを指定している。

table = instance.table('hits')

列ファミリーを hits_family_1 に、キーを hits_value に指定している。

column_family_id = 'hits_family_1'
row_key = 'hits_value'.encode()

テーブル書き込み

テーブルからキーをもとに row オブジェクトを生成する。
set_cell() メソッドを呼びだし、指定したキーに対して値を格納する。
mutate_rows() メソッドを呼び出し、複数の行を一括で変更する（1行ごとの変更メソッドは用意されていないみたい）。

rows = []
row = table.row(row_key)
row.set_cell(column_family_id, row_key, str(new_value), timestamp=datetime.datetime.utcnow())
rows.append(row)
table.mutate_rows(rows)

テーブル読み込み

row_filters.CellsColumnLimitFilter(1) で、読み込むデータにフィルター（最新バージョンのみを返す）をかける。
read_row() メソッドを呼び出し、指定したキーの row オブジェクトを生成する。
cell_values() メソッドを呼び出し、指定したキーの値を取得する。

row_filter = row_filters.CellsColumnLimitFilter(1)
row = table.read_row(row_key, row_filter)
cell_values = row.cell_values(column_family_id, row_key)
for value, timestamp in cell_values:
    count = value.decode('utf-8')

フィルターについて

row_filter = row_filters.CellsColumnLimitFilter(1)

上記のフィルターを使用した場合と使用しない場合では、取得結果に下記のような差異が出る。

フィルターなし

(b'3', 1584944979496000)
(b'2', 1584944978687000)
(b'1', 1584944958173000)

フィルターあり

(b'3', 1584944979496000)

終わりに

Python アプリケーションを使って、Bigtable へアクセスすることができた。

ただ、ほとんどが公式ドキュメントのコピペであり、コードの理解が曖昧なのが心配である。

特にフィルターについて、全くわかっていない…

参考サイト

https://cloud.google.com/bigtable/docs/samples-python-hello?hl=ja

https://qiita.com/yamaneko_usg3/items/b48659967daba80bbecd

https://itnext.io/pubsub-to-bigtable-piping-your-data-stream-in-via-gcp-cloud-functions-a2ef785935b5

https://googleapis.dev/python/bigtable/latest/index.html