BigQueryテーブルのスキーマ比較とHTML差分レポートの作成方法

はじめに

技術記事初心者ですがアドカレ初参加！！！
今回は、BigQueryで2つのテーブルのスキーマ（カラム名、データ型、モード、Description）が一致しているかを比較し、HTML形式の差分レポートを作成してみたいと思います。

背景と目的

BigQueryでデータを扱う際、開発環境やテスト環境で作成したテーブルが、本番環境のテーブルと同じスキーマを持っているか確認したい場合があります。特に、お試しでコンソールとかで作ったものと実際SQLで作ったテーブルの定義が異なることがちらほら、、、
ちょっとまだ考慮たりてないとことかあるけど基盤レベルになれば嬉しいです

解決策の概要

• PythonとGoogle Cloud BigQueryライブラリを使用して、テーブルのスキーマ情報を取得します。
• スキーマをフラット化し、JSON文字列に変換します。
• difflibモジュールを使用して、2つのスキーマ間の差分を比較し、HTMLファイルとして出力します。

必要な環境と準備

以下のPythonライブラリを使用します。(権限周りの設定がめんどくさいのでCloud Shell上で動かす想定)

• google-cloud-bigquery

• json(標準ライブラリ)

• difflib(標準ライブラリ)

  pip install google-cloud-bigquery
  
  

お試し版コード

import json
import difflib
from google.cloud import bigquery

client = bigquery.Client()

# テーブルの参照を作成
correct_table_ref = client.dataset('test').table('test1')
test_table_ref = client.dataset('test').table('test2')

# テーブルのメタデータを取得
correct_table = client.get_table(correct_table_ref)
test_table = client.get_table(test_table_ref)

# スキーマを整形する関数
def flatten_schema(schema, prefix=''):
    fields = {}
    for field in schema:
        field_name = f"{prefix}{field.name}"
        fields[field_name] = {
            'type': field.field_type,
            'mode': field.mode,
            'description': field.description or ''
        }
    return fields

# 正解のテーブルとテスト用のテーブルのスキーマを整形
correct_fields = flatten_schema(correct_table.schema)
test_fields = flatten_schema(test_table.schema)

# 整形したスキーマをJSON文字列に変換
def fields_to_json(fields):
    # フィールド名でソート
    sorted_fields = dict(sorted(fields.items()))
    return json.dumps(sorted_fields, indent=2, ensure_ascii=False)

correct_schema_json = fields_to_json(correct_fields)
test_schema_json = fields_to_json(test_fields)

# 差分を生成
diff = difflib.HtmlDiff().make_file(
    correct_schema_json.splitlines(),
    test_schema_json.splitlines(),
    fromdesc='正解のテーブル',
    todesc='テスト用のテーブル',
    context=False,
    numlines=5
)

# HTMLファイルに出力
with open('schema_diff.html', 'w', encoding='utf-8') as f:
    f.write(diff)

print("差分が'schema_diff.html'に出力されました。ブラウザで開いて確認してください。")

スクリプトの説明

1.対象のテーブルのスキーマ情報を取得する

correct_table_ref = client.dataset('test').table('test1')
test_table_ref = client.dataset('test').table('test2')

# テーブルのメタデータを取得
correct_table = client.get_table(correct_table_ref)
test_table = client.get_table(test_table_ref)

print(correct_table.schema)

↓みたいなのが返ってくる

[SchemaField('column1', 'INTEGER', 'REQUIRED', None, 'column1', (), None), SchemaField('column2', 'INTEGER', 'REQUIRED', None, 'column2', (), None), SchemaField('column3', 'INTEGER', 'REQUIRED', None, 'column3', (), None), SchemaField('column4', 'INTEGER', 'REQUIRED', None, 'column4', (), None), SchemaField('column5', 'INTEGER', 'REQUIRED', None, 'column5', (), None), SchemaField('column6', 'STRING', 'REQUIRED', None, 'column6', (), None), SchemaField('column7', 'INTEGER', 'REQUIRED', None, 'column7', (), None)]

2.取得した結果から比較したいものを抽出する

• flatten_schema関数で確認したい項目を抽出します。

# スキーマを整形する関数
def flatten_schema(schema, prefix=''):
    fields = {}
    for field in schema:
        field_name = f"{prefix}{field.name}"
        fields[field_name] = {
            'type': field.field_type,
            'mode': field.mode,
            'description': field.description or ''
        }
    return fields

# 正解のテーブルとテスト用のテーブルのスキーマを整形
correct_fields = flatten_schema(correct_table.schema)
test_fields = flatten_schema(test_table.schema)

print(correct_fields)

↓みたいなのが返ってくる

{'column1': {'type': 'INTEGER', 'mode': 'REQUIRED', 'description': 'column1'}, 'column2': {'type': 'INTEGER', 'mode': 'REQUIRED', 'description': 'column2'}, 'column3': {'type': 'INTEGER', 'mode': 'REQUIRED', 'description': 'column3'}, 'column4': {'type': 'INTEGER', 'mode': 'REQUIRED', 'description': 'column4'}, 'column5': {'type': 'INTEGER', 'mode': 'REQUIRED', 'description': 'column5'}, 'column6': {'type': 'STRING', 'mode': 'REQUIRED', 'description': 'column6'}, 'column7': {'type': 'INTEGER', 'mode': 'REQUIRED', 'description': 'column7'}}

3.抽出結果のソート

• fields_to_json関数で、スキーマ情報をソートしてJSON形式に変換します。この段階でソートすることで、フィールドの順序に依存しない比較が可能になります。

def fields_to_json(fields):
    # フィールド名でソート
    sorted_fields = dict(sorted(fields.items()))
    return json.dumps(sorted_fields, indent=2, ensure_ascii=False)

correct_schema_json = fields_to_json(correct_fields)
test_schema_json = fields_to_json(test_fields)

print(correct_fields)

4.HTMLレポートの出力

• difflib.HtmlDiffを使用して、2つのJSONを比較し、HTML形式の差分レポートを生成します。

# 差分を生成
diff = difflib.HtmlDiff().make_file(
    correct_schema_json.splitlines(),
    test_schema_json.splitlines(),
    fromdesc='正解のテーブル',
    todesc='テスト用のテーブル',
    context=False,
    numlines=5
)

# HTMLファイルに出力
with open('schema_diff.html', 'w', encoding='utf-8') as f:
    f.write(diff)

print("差分が'schema_diff.html'に出力されました。ブラウザで開いて確認してください。")

実行結果

比較対象テーブル

出力された比較結果

最後に

なんかスキーマの比較って実際どうやんのがいいのかわからんけどINFORMATION_SCHEMAだけだとやりずらい？？
DDLの項目があるからDDL同士で比較？？