Bigqueryを用いた開発で気を付けるべきことや考慮すべきこと、DynamoDBのデータ取得時の注意点を解説

はじめに

この記事ではBigQueryとDynamoDBを使用する際のテクニカルなTipsを紹介します。

また、開発するうえで不便だったこと、考えておくべきことなども説明しています。

BigQueryへのデータの追加

BigQueryにデータを追加する際に、insert_rowとload_table_fromという二つの方法が存在します。

insert_rowには追加できる行数に制限があるため大量のデータを扱うときはload_table_from一択です。 BigQueryにはjobという非同期で処理を行ってくれる機能があり、これを利用すると大きな処理でもタイムアウトが発生せず実装できます。

load_table_fromを使用すると以下のようになります。

def bq_load_from_gcs(gcs_uri, project, dataset, table):
    """Execute BQ load from GCS"""
    client = bigquery.Client(project)
    job_config = bigquery.LoadJobConfig()
    job_config.write_disposition = bigquery.WriteDisposition.WRITE_TRUNCATE
    job_config.source_format = bigquery.SourceFormat.NEWLINE_DELIMITED_JSON
    load_job = client.load_table_from_uri(
        gcs_uri,
        client.dataset(dataset).table(table),
        job_config=job_config,
    )

カスタム定義関数

BigQueryにはUDFという独自関数を作ることのできる機能があります。関数化することによって、コードを再利用することができることやといったメリットを享受することができます。また、SQLだけでなく、JavaScriptで関数を定義することもでき、SQLだけでは実現できない処理を行うことができます。

CREATE TEMP FUNCTION add_numbers(x FLOAT64, y FLOAT64)
RETURNS FLOAT64
LANGUAGE js AS """
  return x + y;
""";

ScheduledQuery関連

BigQueryでQueryのテストは簡単に行うことができますが、テーブルの値を実際に変更するScheduledQueryのテストを行う際、毎回スケジュールを変更して行っていました。しかし、実際にはバックフィル実行という便利な機能があり、スケジュールの変更を行わずにテストを行うことが可能です。

弊社ではGithub上でコードを管理し、レビューの依頼を行ったりしていますが、BigQueryのscheduled queryではそのようなことができません。また、バージョン管理ができないため、ロールバックしたいといったことが簡単にできずに非常に不便でした。

小規模の開発ならそこまで問題はありませんが、大規模なデータ更新パイプラインの場合は以下のような方法を検討することをおすすめします。

IOとロジックの分離

以下のようなjsonをgcsに保存する処理などはほかのプロジェクトでも使いまわすことが多いため、関数化しておくと非常に便利です。また、読みだしたデータに対して処理を行いたいとなった場合に、この関数に処理を追加してしまうと関数が複雑化してしまいます。

関数名からこの関数が、jsonをgcsに保存する関数であると判断して使用したら、実は出力されたデータが操作されていると非常に使いづらいです。また、このような処理はほかの部分でも使えるように、関数の中でロジックとIOを分離しておくと、関数の再利用性が高まります。

def _write_json_to_gcs(bucket, filename, rows, write_mode):
    """Write JSON/JSONL into GCS. write_mode is json or jsonl"""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket)
    blob = bucket.blob(filename)
    with blob.open("w") as f:
        if write_mode == "jsonl":
            for row in rows:
                f.write(json.dumps(row) + "\n")
        elif write_mode == "json":
            f.write(json.dumps(rows))

Dynamodbのscanについて

Dynamodbのscanは最大1MBまでのScanしか対応していないため、1MBを超える場合は、次のページがあるかどうかを確認して、再度scanを行う必要があります。以下のコードは、Dynamodbのscanを行う際に、次のページがあるかどうかを確認して、再度scanを行う処理を行っています。

count = 1
while "LastEvaluatedKey" in response:
    response = dynamodb.scan(
        TableName=dynamo_table_name,
        ExclusiveStartKey=response["LastEvaluatedKey"],
    )
    file_name = f"{folder_name}/{count}.json"
    data = pre_proceed(response["Items"], file_name)
    write_json_to_gcs(GCS_BUCKET, file_name, data, "jsonl")
    count += 1

参考

ScanでDynamoDBテーブルから1MB以上のデータを取得する方法について

備考

Hakky ではエンジニアを募集中です！まずは話してみたいなどでも構いませんので、ぜひお気軽に採用ページからお問い合わせくださいませ。