こんばんは。佐藤です👓

案件でGoogle Cloudの「Dataplex Universal Catalog」を利用することがありましたので、どのように活用したのか、備忘録的に残したいと思います。

Dataplex Universal Catalogとは

Google Cloudが提供しているサービスの一つで、Google Cloud専用のデータカタログです。(以下 Dataplexと表記)

Google Cloud内に格納されているデータに対して、メタデータを付与したり、品質チェックなどの管理や統合したりすることが可能です。

導入のきっかけ

BigQueryを中心にデータ活用を進める中で、分析用・業務用のテーブルやデータセットが継続的に増加していきました。一方で、開発に関わるチームが増えるにつれて「誰が・どのテーブルに・どんな意味のデータを入れているのか」が分かりにくくなり、必要なデータに辿り着くまでのコストが高くなっていきました。

具体的には、次のような課題が目立つようになりました。

テーブル数の増加により、目的のデータがどこにあるか探索に時間がかかる
命名や定義がチームごとに異なり、同じようなテーブルが複数存在してしまう
データの意味や更新頻度、責任者（オーナー）が分からず、利用の判断が難しい
一部の詳しい人に問い合わせが集中し、データ利用が属人化する

こうした状況を改善し、より多くの人が安心してデータを見つけて使える状態（いわゆる「データの民主化」）を進めるために、メタデータの整備や品質管理を一元的に扱える仕組みが必要になりました。

そこで、Google Cloud上のデータに対してカタログ化・検索性向上・ガバナンス強化をまとめて実現できる選択肢として、Dataplexの導入を検討しました。

メタデータ

データの民主化のためまず考えたことは、各テーブルに対してメタデータを付与することでした。

Dataplexではメタデータを付与する方法として、アスペクトというものが存在します。

docs.cloud.google.com

データにアスペクト(メタデータ)を付与すると、Dataplexの画面で検索することができるイメージです。

また、アスペクトの中にも複数の情報を持たせることが可能です。
文字列や整数型、配列、ブール値など様々な型を設定することができます。

アスペクト設計

テーブルに対して付与したメタデータの一例を共有します。

例)全てのテーブルに対して共通で付与したアスペクト(common)

commonアスペクト

アスペクト値	具体値
データ加工状態	source, cleansing, staging, integration, mart
データ基盤元	Google Cloud Storage, S3…
データ出力先	Google Cloud Storage, S3…
マスタフラグ	マスタならTrue,トランザクションならFalse
データ管理者	ベンダ名など
情報レベル	公開情報,秘密情報など

共通項目としてデータ管理者や情報レベルをメタデータに付与することで、データに不具合が起きた際の問い合わせ先が明確になり、機密性に応じた参照範囲や取り扱いルールも判断しやすくなります。

私の所属部署ではCDP構築を主に専門としており、加工状態が不明だと「どのデータを正とすべきか」「どの工程のデータを連携すべきか」の判断が困難になります。そのため、source / staging / mart などの加工状態をアスペクトとして明示することで、データの位置づけを統一した運用が可能になり、連携や設計の手戻りを減らすことを目的としています。

アスペクト検索の罠~AND検索難しい問題~

何だこのタイトルはと思うかもしれませんが、異なるアスペクト間のAND検索に少しコツが必要です。

例えば、AアスペクトとBアスペクトを持つテーブルを探したい時、GUI上のアスペクト一覧画面から２つチェックすれば良いだろう、と考えると思います。

しかしDataplexは『なぜかOR検索になる』仕様となっています。

例)　test1というテーブルにtest_aspectを設定

test1テーブルにはtest_aspectのみ紐づいているはずですが、他のアスペクトを選択しても検索画面に残ったままです。

AND検索ができないというわけではないのです。

一応検索バーが存在しており、例えばAアスペクトとBアスペクトをANDで検索したければ以下のように検索バーに入力することで検索できます。

詳しい検索構文はこちら↓

docs.cloud.google.com

ただ、Google Cloud製品にあまり触れていないユーザにとってはなかなか検索するのにハードルがある気がしますね…

ここは改善して欲しいポイントだなと思います。

アスペクト付与

この章ではDataplex上で作成したアスペクトをどのようにしてテーブルに付与したのかご紹介します。

アスペクト付与方法

アスペクトを手動で付与するのはさすがに大変なので、Pythonでメタデータを記載したデータを参照してDataplex上に付与する方法を利用しました。

以下は、CSVデータにメタデータが格納されている想定でアスペクトを付与するコード例です。

import pandas as pdimport requestsimport subprocess
import requests
import subprocess

# CSV読み込み
df = pd.read_csv("test.csv")

# GCPアクセストークン取得
 def get_access_token():
     return subprocess.check_output(['gcloud', 'auth', 'print-access-token']).decode().strip()

#access_token = get_access_token()

# 固定カラム名
fixed_cols = ['TABLE_NAME', 'PROJECT_ID', 'DATASET_ID', 'ASPECT_NAME', 'REGION']

for idx, row in df.iterrows():
    table_name = row['TABLE_NAME']
    project_id = row['PROJECT_ID']
    dataset_id = row['DATASET_ID']
    aspect_name = row['ASPECT_NAME']
    region = row['REGION']

    entry_id = f"bigquery.googleapis.com/projects/{project_id}/datasets/{dataset_id}/tables/{table_name}"
    aspect_type_full_name = f"projects/{project_id}/locations/{region}/aspectTypes/{aspect_name}"
    aspect_key = f"{project_id}.{region}.{aspect_name}"

    # 固定カラム以外を data として抽出
    data_dict = {col: row[col] for col in df.columns if col not in fixed_cols and pd.notnull(row[col])}

    payload = {
        "aspects": {
            aspect_key: {
                "aspectType": aspect_type_full_name,
                "data": data_dict if data_dict else None  # データがなければ None
            }
        }
    }

    headers = {
        "Authorization": f"Bearer hogehoge",
        "Content-Type": "application/json"
    }

    print(payload)
    response = requests.patch(url, json=payload, headers=headers)
    print(f"{table_name}: {response.status_code}, {response.text}")

アスペクト付与の落とし穴

ただ、これを一回実行すればOKというわけではありません。 Dataformなどでテーブルの洗い替え処理を入れている場合、洗い替えされるたびにアスペクトは消えてしまいます。

(Cloud functionsなどで洗い替え処理をしている場合も同様)

そのため、洗い替えされるたびにアスペクトを付与するロジックが必要となります。

それを踏まえて考えたシステム構成図が以下となります。

システム構成図

①でCloud loggingのシンクを利用し、Dataformの洗い替えキャッチ→pubsub→Workflows→Cloud Functionsという流れで洗い替えが起きるたびにその洗い替え対象のテーブルをCloud Loggingから拾って再度付与するという流れで処理を行っています。

注意点は以下の通りです。

API制限に注意！
- １分あたり1,500リクエストまでなので、多数の洗い替え処理している環境では実行速度の制限など設けましょう。
- https://docs.cloud.google.com/dataplex/docs/quotas?hl=ja
スプレッドシートは外部テーブルではなく、物理テーブルとして持つ
- 外部テーブルには参照の制限があります。物理テーブルにしておけばクエリをCloud Functions上で何回読んでも耐えることが可能なため、特に理由がなければ物理テーブルにしましょう。
- https://docs.cloud.google.com/bigquery/docs/external-tables?hl=ja