0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@satoshi_enomoto(Satoshi ENOMOTO)

BigQuery テーブルに対して Dataplex API でデータプロファイルとデータ品質を実施する手順

0
Last updated at Posted at 2026-07-28

はじめに

Google Cloud では、Knowledge Catalog (旧 Dataplex) の DataScan を利用することで、BigQuery テーブルに対するデータプロファイリングやデータ品質チェックを実行できます。

データプロファイルでは、各列の NULL 率、値の分布、一意性、最小値・最大値など、データの特徴を自動的に確認できます。一方、データ品質スキャンでは、NULL 禁止、一意性、値の範囲といったルールを定義し、対象データが期待する品質基準を満たしているかを判定できます。

本記事では、Python の google-cloud-dataplex を利用し、次の内容を確認します。

  • データプロファイル スキャンの作成と実行
  • スキャン結果の BigQuery テーブルへの出力
  • データプロファイル結果の Knowledge Catalog への公開
  • データ品質ルールの作成とスキャンの実行

事前準備

共通設定

以降の処理で使用するプロジェクト ID、BigQuery のロケーション、データセット名、テーブル名、スキャン ID を変数として定義します。

共通設定 
PROJECT_ID = "<PROJECT_ID>"
BQ_LOCATION = "<BQ_LOCATION>"
DATASET_ID = "<DATASET_ID>"
SOURCE_TABLE_ID = "<SOURCE_TABLE_NAME>"
PROFILE_RESULT_TABLE_ID = "<PROFILE_RESULT_TABLE_NAME>"
QUALITY_RESULT_TABLE_ID = "<QUALITY_RESULT_TABLE_NAME>"
DATA_PROFILE_SCAN_ID = "<DATA_PROFILE_SCAN_ID>"
DATA_QUALITY_SCAN_ID = "<DATA_QUALITY_SCAN_ID>"

SOURCE_TABLE = f"{PROJECT_ID}.{DATASET_ID}.{SOURCE_TABLE_ID}"
PROFILE_RESULT_TABLE = f"{PROJECT_ID}.{DATASET_ID}.{PROFILE_RESULT_TABLE_ID}"
QUALITY_RESULT_TABLE = f"{PROJECT_ID}.{DATASET_ID}.{QUALITY_RESULT_TABLE_ID}"

BigQuery データセットの作成

BigQuery データセットの作成 
from google.cloud import bigquery

bq_client = bigquery.Client(project=PROJECT_ID)

dataset_ref = bigquery.Dataset(f"{PROJECT_ID}.{DATASET_ID}")
dataset_ref.location = BQ_LOCATION

dataset = bq_client.create_dataset(
    dataset_ref,
    exists_ok=True,
)

print(f"データセットを確認しました: {dataset.full_dataset_id}")
print(f"ロケーション: {dataset.location}")

サンプル テーブルの作成

データプロファイルとデータ品質スキャンの動作を確認するため、顧客情報を想定したサンプルテーブルを作成します。NULL を含む列を用意することで、列ごとの統計情報や品質ルールの判定結果を確認しやすくします。

サンプル テーブルの作成 
create_table_sql = f"""
CREATE OR REPLACE TABLE `{SOURCE_TABLE}`  AS
SELECT
  1 AS customer_id,
  '田中 太郎' AS customer_name,
  'tokyo@example.com' AS email,
  35 AS age,
  120000 AS annual_purchase_amount,
  DATE '2026-01-10' AS registered_date,
  TIMESTAMP '2026-07-01 01:00:00+00' AS updated_at
UNION ALL
SELECT
  2,
  '佐藤 花子',
  'sato@example.com',
  28,
  85000,
  DATE '2026-02-15',
  TIMESTAMP '2026-07-02 02:00:00+00'
UNION ALL
SELECT
  3,
  '鈴木 一郎',
  NULL,
  42,
  210000,
  DATE '2026-03-20',
  TIMESTAMP '2026-07-03 03:00:00+00'
UNION ALL
SELECT
  4,
  '高橋 美咲',
  'takahashi@example.com',
  NULL,
  43000,
  DATE '2026-04-05',
  TIMESTAMP '2026-07-04 04:00:00+00'
UNION ALL
SELECT
  5,
  '山田 次郎',
  'yamada@example.com',
  35,
  120000,
  DATE '2026-05-18',
  TIMESTAMP '2026-07-05 05:00:00+00';
"""

query_job = bq_client.query(
    create_table_sql,
    location=BQ_LOCATION,
)
query_job.result()

print(f"サンプルテーブルを作成しました: {SOURCE_TABLE}")

データプロファイル

データプロファイル スキャンの作成

Dataplex API を利用して、サンプルテーブルを対象としたデータプロファイルスキャンを作成します。スキャンはオンデマンド実行とし、実行結果は指定した BigQuery テーブルへ出力するよう設定します。

データプロファイル スキャンの作成 
from google.api_core.client_options import ClientOptions
from google.cloud import dataplex_v1


client = dataplex_v1.DataScanServiceClient(
    transport="rest",
)


parent = f"projects/{PROJECT_ID}/locations/{BQ_LOCATION}"

source_table = (
    f"//bigquery.googleapis.com/projects/{PROJECT_ID}"
    f"/datasets/{DATASET_ID}/tables/{SOURCE_TABLE_ID}"
)

result_table = (
    f"//bigquery.googleapis.com/projects/{PROJECT_ID}"
    f"/datasets/{DATASET_ID}/tables/{PROFILE_RESULT_TABLE_ID}"
)


data_scan = dataplex_v1.DataScan(
    display_name="Customer data profile scan",
    description="Pythonから作成したBigQueryデータプロファイルスキャン",
    data=dataplex_v1.DataSource(
        resource=source_table
    ),
    data_profile_spec=dataplex_v1.DataProfileSpec(
        post_scan_actions=dataplex_v1.DataProfileSpec.PostScanActions(
            bigquery_export=(
                dataplex_v1.DataProfileSpec.PostScanActions.BigQueryExport(
                    results_table=result_table
                )
            )
        ),
    ),
    execution_spec=dataplex_v1.DataScan.ExecutionSpec(
        trigger=dataplex_v1.Trigger(
            on_demand=dataplex_v1.Trigger.OnDemand()
        )
    ),
)


request = dataplex_v1.CreateDataScanRequest(
    parent=parent,
    data_scan=data_scan,
    data_scan_id=f"{DATA_PROFILE_SCAN_ID}",
)

operation = client.create_data_scan(request=request)

print(operation.operation.name)

データプロファイル スキャンの実行

作成したデータプロファイルスキャンをオンデマンドで実行します。run_data_scan() の戻り値からジョブ名を取得することで、後から実行状態やスキャン結果を確認できます。

データプロファイル スキャンの実行 
from google.cloud import dataplex_v1


client = dataplex_v1.DataScanServiceClient(
    transport="rest",
)


scan_name = (
    f"projects/{PROJECT_ID}"
    f"/locations/{BQ_LOCATION}"
    f"/dataScans/{DATA_PROFILE_SCAN_ID}"
)

response = client.run_data_scan(
    request=dataplex_v1.RunDataScanRequest(
        name=scan_name
    )
)


job_name = response.job.name

print("スキャンを開始しました")
print(f"job_name: {job_name}")

スキャン結果の確認

データプロファイルスキャンの結果を、出力先として指定した BigQuery テーブルから確認します。列ごとの NULL 率、値の分布、一意性、最小値・最大値などを確認できます。

スキャン結果の確認 
df = bq_client.query(
    f"""
    SELECT *
    FROM `{PROFILE_RESULT_TABLE}`
    ORDER BY job_start_time DESC;
    """,
    location=BQ_LOCATION,
).to_dataframe()

df

image.png

また、Google Cloud コンソールの BigQuery にある「メタデータのキュレーション」画面からも、スキャン定義や実行結果を確認できます。

image.png

Knowledge Catalog への公開

データプロファイル結果を Knowledge Catalog に公開すると、対象となる BigQuery テーブルの詳細画面から最新のプロファイル情報を確認できます。

既存のデータプロファイル スキャンに対して catalog_publishing_enabled を有効化し、再度スキャンを実行することで、Knowledge Catalog および BigQuery テーブルの「データプロファイル」タブに結果を表示します。

データプロファイル スキャンの更新 
from google.cloud import dataplex_v1
from google.protobuf.field_mask_pb2 import FieldMask


scan_name = (
    f"projects/{PROJECT_ID}"
    f"/locations/{BQ_LOCATION}"
    f"/dataScans/{DATA_PROFILE_SCAN_ID}"
)


client = dataplex_v1.DataScanServiceClient(
    transport="rest"
)


data_scan = dataplex_v1.DataScan(
    name=scan_name,
    data_profile_spec=dataplex_v1.DataProfileSpec(
        catalog_publishing_enabled=True,
    ),
)

request = dataplex_v1.UpdateDataScanRequest(
    data_scan=data_scan,
    update_mask=FieldMask(
        paths=[
            "data_profile_spec.catalog_publishing_enabled",
        ]
    ),
)


operation = client.update_data_scan(request=request)

print(operation.operation.name)

データプロファイル スキャンの構成を確認すると、Knowledge Catalog に結果を公開する が 〇 になっています。

image.png

再度、データプロファイルスキャンを実行すると、Knowledge Catalog にスキャン結果が表示されます。

image.png

また、BigQuery テーブルのデータプロファイル タブにも結果が表示されるようになります。

image.png

データ品質

テーブル品質スキャンの作成

Dataplex API を利用して、サンプルテーブルを対象としたデータ品質スキャンを作成します。

今回は、customer_id が NULL でないこと、customer_id が一意であること、age が 0 から 120 の範囲内であることを品質ルールとして定義します。スキャン結果は BigQuery テーブルへ出力し、あわせて Knowledge Catalog への公開も有効化します。

テーブル品質スキャンの作成 
from google.cloud import dataplex_v1


client = dataplex_v1.DataScanServiceClient(
    transport="rest"
)


parent = f"projects/{PROJECT_ID}/locations/{BQ_LOCATION}"

source_table = (
    f"//bigquery.googleapis.com/projects/{PROJECT_ID}"
    f"/datasets/{DATASET_ID}"
    f"/tables/{SOURCE_TABLE_ID}"
)

result_table = (
    f"//bigquery.googleapis.com/projects/{PROJECT_ID}"
    f"/datasets/{DATASET_ID}/tables/{QUALITY_RESULT_TABLE_ID}"
)

rules = [
    # customer_idがNULLでないこと
    dataplex_v1.DataQualityRule(
        name="customer-id-not-null",
        column="customer_id",
        dimension="COMPLETENESS",
        threshold=1.0,
        non_null_expectation=(
            dataplex_v1.DataQualityRule.NonNullExpectation()
        ),
    ),

    # customer_idが一意であること
    dataplex_v1.DataQualityRule(
        name="customer-id-unique",
        column="customer_id",
        dimension="UNIQUENESS",
        uniqueness_expectation=(
            dataplex_v1.DataQualityRule.UniquenessExpectation()
        ),
    ),

    # ageが0～120の範囲内であること
    dataplex_v1.DataQualityRule(
        name="age-valid-range",
        column="age",
        dimension="VALIDITY",
        threshold=1.0,
        ignore_null=True,
        range_expectation=(
            dataplex_v1.DataQualityRule.RangeExpectation(
                min_value="0",
                max_value="120",
            )
        ),
    ),
]

data_scan = dataplex_v1.DataScan(
    display_name="Customer data quality scan",
    description="Pythonから作成したデータ品質スキャン",
    data=dataplex_v1.DataSource(
        resource=source_table
    ),
    data_quality_spec=dataplex_v1.DataQualitySpec(
        rules=rules,
        catalog_publishing_enabled=True,
        post_scan_actions=dataplex_v1.DataQualitySpec.PostScanActions(
            bigquery_export=(
                dataplex_v1.DataQualitySpec.PostScanActions.BigQueryExport(
                    results_table=result_table
                )
            )
        ),
    ),
    execution_spec=dataplex_v1.DataScan.ExecutionSpec(
        trigger=dataplex_v1.Trigger(
            on_demand=dataplex_v1.Trigger.OnDemand()
        )
    ),
)


request = dataplex_v1.CreateDataScanRequest(
    parent=parent,
    data_scan=data_scan,
    data_scan_id=DATA_QUALITY_SCAN_ID,
)

operation = client.create_data_scan(request=request)

print(operation.operation.name)

データ品質スキャンの実行

作成したデータ品質スキャンをオンデマンドで実行します。実行後に取得できるジョブ名を利用することで、API からジョブの状態や品質スコア、ルールごとの判定結果を確認できます。

データ品質スキャンの実行 
from google.cloud import dataplex_v1


client = dataplex_v1.DataScanServiceClient(
    transport="rest"
)


scan_name = (
    f"projects/{PROJECT_ID}"
    f"/locations/{BQ_LOCATION}"
    f"/dataScans/{DATA_QUALITY_SCAN_ID}"
)


run_response = client.run_data_scan(
    request=dataplex_v1.RunDataScanRequest(
        name=scan_name
    )
)

job_name = run_response.job.name

print("スキャンを開始しました")
print(f"job_name: {job_name}")

スキャン結果の確認

データ品質スキャンの結果を、指定した BigQuery の結果テーブルから確認します。結果には、スキャン全体の合否、品質スコア、対象行数、ルールごとの合否や評価結果などが記録されます。

スキャン結果の確認 
df = bq_client.query(
    f"""
    SELECT *
    FROM `{QUALITY_RESULT_TABLE}`
    ORDER BY job_start_time DESC;
    """,
    location=BQ_LOCATION,
).to_dataframe()

df

image.png

Google Cloud コンソールの BigQuery にある「メタデータのキュレーション」画面からも、データ品質結果を確認できます。

image.png

また、Knowledge Catalog および BigQuery テーブルの「データプロファイル」タブにも結果が表示されます。

image.png

image.png

データ品質での NULL の取り扱いについて

サンプルデータの age 列には NULL が含まれていますが、品質は合格となっています。これは age の範囲チェックのルールに ignore_null=True 指定しているため、チャックの対象から除外しているためです。

NULL をチェック対象とするには、ignore_null=False とするか、もしくは NULL禁止は別のルール NonNullExpectation で定義します。

rules = [
    # ageがNULLでないこと
    dataplex_v1.DataQualityRule(
        name="age-not-null",
        column="age",
        dimension="COMPLETENESS",
        threshold=1.0,
        non_null_expectation=(
            dataplex_v1.DataQualityRule.NonNullExpectation()
        ),
    ),

    # ageが0～120であること (ignore_null=True のまま)
    dataplex_v1.DataQualityRule(
        name="age-valid-range",
        column="age",
        dimension="VALIDITY",
        threshold=1.0,
        ignore_null=True,
        range_expectation=(
            dataplex_v1.DataQualityRule.RangeExpectation(
                min_value="0",
                max_value="120",
            )
        ),
    ),
]

Lakehouse for Apache Iceberg への実施

Lakehouse for Apache Iceberg に対しても実施可能でした。

image.png

image.png

image.png

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?