BigQuery のテーブルを Parquet で GCS へエクスポートして Databricks で読み込む際の動作検証

はじめに

本記事では、BigQuery のテーブルデータを Parquet 形式で Google Cloud Storage（GCS）へエクスポートし、Azure Databricks から直接読み込んで Delta テーブルへ反映するまでの動作を検証します。

検証では、BigQuery の EXPORT DATA を使用して、updated_at を基準に対象データを抽出し、エクスポート実行時刻を付与した Parquet ファイルを GCS へ出力します。その後、Azure Databricks のクラシックコンピュートに GCS の認証情報を設定し、Spark から gs:// パスを直接参照します。

最後に、読み込んだ Parquet データの型を Databricks 側のテーブル定義に合わせて変換し、Delta Lake の replaceWhere を使用して対象期間のみを選択的に上書きします。

今回の構成は以下のとおりです。

BigQuery
   |
   | EXPORT DATA（Parquet）
   v
Google Cloud Storage
   |
   | GCS Connector / gs://
   v
Azure Databricks クラシックコンピュート
   |
   | spark.read.parquet
   v
Spark DataFrame
   |
   | replaceWhere
   v
Delta テーブル

GCS の環境構築

GCS の環境構築については以下の manabian 氏の記事をご参照ください。

サービスアカウントの JSON キー作成時のエラー
サービスアカウントの JSON キー作成時に以下のようなエラーが発生しました。

サービス アカウント キーの作成をブロックする組織ポリシーが組織に適用されています。

こちらは次の組織ポリシーを非アクティブにすることで解決しました。
iam.disableServiceAccountKeyCreation

検証用の BigQuery テーブルを作成

BigQuery に検証用テーブルを作成します。

-- サンプルテーブル作成
CREATE OR REPLACE TABLE abc.export_demo_table
(
  sale_id       INT64,
  product_name  STRING,
  quantity      INT64,
  unit_price    NUMERIC,
  sale_date     DATE,
  updated_at    TIMESTAMP
);

-- テストデータ挿入
INSERT INTO abc.export_demo_table
(
  sale_id,
  product_name,
  quantity,
  unit_price,
  sale_date,
  updated_at
)
VALUES
  (
    1,
    'Notebook',
    2,
    500.00,
    DATE '2026-07-20',
    TIMESTAMP '2026-07-20 01:00:00+00'
  ),
  (
    2,
    'Pen',
    10,
    120.00,
    DATE '2026-07-20',
    TIMESTAMP '2026-07-20 02:00:00+00'
  ),
  (
    3,
    'File',
    5,
    300.00,
    DATE '2026-07-21',
    TIMESTAMP '2026-07-21 01:00:00+00'
  ),
  (
    4,
    'Desk',
    1,
    25000.00,
    DATE '2026-07-21',
    TIMESTAMP '2026-07-21 02:00:00+00'
  );

SELECT * FROM abc.export_demo_table;

Parquet で GCS へエクスポート

updated_at の期間を指定してエクスポートします。また、export_timestamp を付与します。

DECLARE export_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP();
DECLARE export_uri STRING;
DECLARE export_from TIMESTAMP DEFAULT '2026-07-21 00:00:00+00';

-- 出力先URIを生成
SET export_uri = FORMAT(
  'gs://export_demo_bucket/selectivity_export/export_demo_table/export_timestamp=%s/export_demo_table_*.parquet',
  FORMAT_TIMESTAMP(
    '%Y-%m-%d %H:%M:%S%Ez',
    export_timestamp,
    'Asia/Tokyo'
  )
);

-- いつから出力するか設定
SET export_from = TIMESTAMP(
  DATETIME(
    CURRENT_DATE(),
    TIME '00:00:00'
  )
);

-- 出力
EXECUTE IMMEDIATE FORMAT("""
  EXPORT DATA OPTIONS (
    uri = '%s',
    format = 'PARQUET',
    overwrite = TRUE,
    compression = 'SNAPPY'
  )
  AS
  SELECT 
    sale_id,
    product_name,
    quantity,
    unit_price,
    sale_date,
    updated_at,
    %T AS export_timestamp
  FROM abc.export_demo_table
  WHERE updated_at >= %T
  """, 
  export_uri,
  export_timestamp,
  export_from
);

Azure Databricks から GCS を操作

Spark構成 にサービスアカウントの認証情報を下記のように設定してクラスターを作成します。

spark.hadoop.google.cloud.auth.service.account.enable true
spark.hadoop.fs.gs.auth.service.account.email {client_email の値}
spark.hadoop.fs.gs.project.id {project_id の値}
spark.hadoop.fs.gs.auth.service.account.private.key {private_key の値}
spark.hadoop.fs.gs.auth.service.account.private.key.id {private_key_id の値}

作成したクラスターで、以下のように読み込むことができることを確認できました。

df_from_gcs = spark.read.parquet("gs://<bucket_mame>/<path>")
df_from_gcs.display()

データ型のマップ
BigQuery から Spark へのデータ型マッピングは以下のようになるようです。

https://learn.microsoft.com/ja-jp/azure/databricks/query-federation/bigquery#data-type-mappings

Azure Databricks で選択的上書きの実行

Databricks 側でサンプルテーブルと初期データを用意します。

%sql
CREATE OR REPLACE TABLE <table_name>
(
  sale_id       BIGINT,
  product_name  STRING,
  quantity      BIGINT,
  unit_price    DECIMAL(18, 2),
  sale_date     DATE,
  updated_at    TIMESTAMP_NTZ,
  export_timestamp TIMESTAMP_NTZ
)
USING DELTA
TBLPROPERTIES (
  'delta.feature.timestampNtz' = 'supported'
);

INSERT OVERWRITE TABLE <table_name>
(
  sale_id,
  product_name,
  quantity,
  unit_price,
  sale_date,
  updated_at,
  export_timestamp
)
VALUES
  (
    1,
    'Notebook',
    2,
    CAST(500.00 AS DECIMAL(18, 2)),
    '2026-07-20',
    '2026-07-20 01:00:00',
    '2026-07-21 01:00:00'
  ),
  (
    2,
    'Pen',
    10,
    CAST(120.00 AS DECIMAL(18, 2)),
    '2026-07-20',
    '2026-07-20 02:00:00',
    '2026-07-21 01:00:00'
  ),
  (
    3,
    'File',
    1,
    CAST(300.00 AS DECIMAL(18, 2)),
    '2026-07-21',
    '2026-07-21 01:00:00',
    '2026-07-21 01:00:00'
  );

SELECT *
FROM <table_name>
ORDER BY sale_id;

GCS から読み込んだ Parquet ファイルを Databricks のテーブルに選択的上書きを実行します。

from pyspark.sql.functions import col

df_from_gcs = spark.read.parquet("gs://<bucket_mame>/<path>")

df_from_gcs_cast = (
    df_from_gcs
    .withColumn(
        "unit_price",
        col("unit_price").cast("decimal(18,2)")
    )
)

(df_from_gcs_cast.write
  .mode("overwrite")
  .option("replaceWhere", "updated_at >= '2026-07-21 00:00:00'")
  .saveAsTable("<table_name>")
)

テーブルが上書きされたことを確認できました。