BigQuery Lakehouse カタログ Iceberg テーブルへの SQL 書き込みとクエリパフォーマンスを検証してみた。

export PROJECT_ID="[PROJECT_ID]"
export LOCATION="asia-northeast1"
export GCS_BUCKET="sb-iceberg-verify-mikami"

bq mk --connection \
  --location="$LOCATION" \
  --connection_type=SPARK \
  --project_id="$PROJECT_ID" \
  "spark-iceberg-verify"

# サービスアカウントを変数に設定
export SPARK_SA=$(bq show --connection --project_id="$PROJECT_ID" \
  --location="$LOCATION" --format=json spark-iceberg-verify \
  | jq -r '.cloudResource.serviceAccountId')

# Spark サービスアカウントへの権限付与
gcloud storage buckets add-iam-policy-binding "gs://$GCS_BUCKET" \
  --member="serviceAccount:$SPARK_SA" --role="roles/storage.objectAdmin"
gcloud projects add-iam-policy-binding "$PROJECT_ID" \
  --member="serviceAccount:$SPARK_SA" --role="roles/biglake.admin"
gcloud projects add-iam-policy-binding "$PROJECT_ID" \
  --member="serviceAccount:$SPARK_SA" --role="roles/serviceusage.serviceUsageConsumer"
gcloud projects add-iam-policy-binding "$PROJECT_ID" \
  --member="serviceAccount:$SPARK_SA" --role="roles/bigquery.dataViewer"
gcloud projects add-iam-policy-binding "$PROJECT_ID" \
  --member="serviceAccount:$SPARK_SA" --role="roles/bigquery.jobUser"

CREATE OR REPLACE TABLE
  iceberg_dataset_tokyo.new_payments
AS
SELECT
  payment_id,
  CASE MOD(payment_id, 3)
    WHEN 0 THEN 'STORE_A' WHEN 1 THEN 'STORE_B' ELSE 'STORE_C'
  END AS store_code,
  (MOD(payment_id, 5) + 1) * 2000 AS amount,
  DATE_ADD(DATE '2026-05-05', INTERVAL (payment_id - 1005) DAY) AS payment_date
FROM UNNEST(GENERATE_ARRAY(1005, 1010)) AS payment_id;

SELECT * FROM iceberg_dataset_tokyo.new_payments
ORDER BY payment_id;

CREATE OR REPLACE PROCEDURE
  iceberg_dataset_tokyo.append_to_iceberg()
WITH CONNECTION `[PROJECT_ID].asia-northeast1.spark-iceberg-verify`
OPTIONS (
  engine = 'SPARK',
  runtime_version = '2.2'
)
LANGUAGE python AS R"""
import subprocess, sys
subprocess.run(
    [sys.executable, "-m", "pip", "install", "pyiceberg[gcsfs,pyarrow]", "--quiet"],
    check=True
)

from pyspark.sql import SparkSession
import google.auth
import google.auth.transport.requests
from pyiceberg.catalog.rest import RestCatalog
from google.cloud import bigquery as bq_lib

creds, _ = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
creds.refresh(google.auth.transport.requests.Request())

spark = SparkSession.builder \
    .appName("append_to_iceberg") \
    .getOrCreate()

# BigQuery Python クライアントで読み込み（Storage Read API 不使用）
bq_client = bq_lib.Client(
    project="[PROJECT_ID]",
    credentials=creds
)
arrow_table = (
    bq_client
    .query(
        'SELECT '
        'CAST(payment_id   AS INT64)  AS payment_id, '
        'CAST(store_code   AS STRING) AS store_code, '
        'CAST(amount       AS INT64)  AS amount, '
        'CAST(payment_date AS DATE)   AS payment_date '
        'FROM `[PROJECT_ID].iceberg_dataset_tokyo.new_payments`'
    )
    .result()
    .to_arrow(create_bqstorage_client=False)
)
print(f"Source rows: {len(arrow_table)}")

catalog = RestCatalog(
    name="lakehouse",
    **{
        "uri": "https://biglake.googleapis.com/iceberg/v1/restcatalog",
        "warehouse": "gs://sb-iceberg-verify-mikami",
        "token": creds.token,
        "header.x-goog-user-project": "[PROJECT_ID]",
        "header.X-Iceberg-Access-Delegation": "vended-credentials",
    }
)

tbl = catalog.load_table("sb_verify.payment_log")
tbl.append(arrow_table)

result = tbl.scan().to_arrow()
print(result)
print(f"Total rows: {len(result)}")
""";

CALL iceberg_dataset_tokyo.append_to_iceberg();

CREATE OR REPLACE TABLE
  iceberg_dataset_tokyo.payment_log_std
AS
SELECT
  payment_id,
  CONCAT('STORE_', LPAD(CAST(MOD(payment_id, 50) + 1 AS STRING), 3, '0')) AS store_code,
  CASE MOD(payment_id, 3) WHEN 0 THEN 'CREDIT' WHEN 1 THEN 'DEBIT' ELSE 'QR' END
    AS payment_method,
  CAST(100 + RAND() * 49900 AS INT64) AS amount,
  DATE_ADD(DATE '2022-01-01', INTERVAL MOD(payment_id, 2000) DAY) AS payment_date
FROM UNNEST(GENERATE_ARRAY(1, 1000000)) AS payment_id;

SELECT COUNT(*) as count, MIN(payment_id) as min_id, MAX(payment_id) as max_id FROM iceberg_dataset_tokyo.payment_log_std;

-- データエクスポート
EXPORT DATA OPTIONS (
  uri = 'gs://sb-iceberg-verify-mikami/perf_test/parquet/*.parquet',
  format = 'PARQUET', overwrite = true
)
AS SELECT * FROM `[PROJECT_ID].iceberg_dataset_tokyo.payment_log_std`;

-- 外部テーブルを作成
CREATE OR REPLACE EXTERNAL TABLE
  `[PROJECT_ID].iceberg_dataset_tokyo.payment_log_ext`
OPTIONS (
  format = 'PARQUET',
  uris   = ['gs://sb-iceberg-verify-mikami/perf_test/parquet/*.parquet']
);

cat > create_perf_table.py << 'EOF'
import os, time
import google.auth, google.auth.transport.requests
import pyarrow as pa, pyarrow.parquet as pq
from google.cloud import storage
from pyiceberg.catalog.rest import RestCatalog

creds, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
creds.refresh(google.auth.transport.requests.Request())

catalog = RestCatalog(
    name="lakehouse",
    **{
        "uri": "https://biglake.googleapis.com/iceberg/v1/restcatalog",
        "warehouse": f"gs://{os.environ['GCS_BUCKET']}",
        "token": creds.token,
        "header.x-goog-user-project": os.environ["PROJECT_ID"],
        "header.X-Iceberg-Access-Delegation": "vended-credentials",
    },
)

# GCS の Parquet ファイルを読み込んで Iceberg テーブルに書き込む
client = storage.Client(project=os.environ["PROJECT_ID"])
blobs = list(client.bucket(os.environ["GCS_BUCKET"]).list_blobs(prefix="perf_test/parquet/"))
print(f"Parquet files: {len(blobs)}")

combined = pa.concat_tables([
    pq.read_table(f"gs://{os.environ['GCS_BUCKET']}/{b.name}") for b in blobs
])
print(f"Total rows: {len(combined)}")

t = catalog.create_table_if_not_exists("sb_verify.payment_log_perf", schema=combined.schema)
t0 = time.perf_counter()
t.append(combined)
print(f"Write elapsed: {time.perf_counter()-t0:.1f}s")
EOF

python3 create_perf_table.py

export PROJECT_ID="[PROJECT_ID]"
export LOCATION="asia-northeast1"
export GCS_BUCKET="sb-iceberg-verify-mikami"

bq mk --connection \
  --location="$LOCATION" \
  --connection_type=CLOUD_RESOURCE \
  --project_id="$PROJECT_ID" \
  bq-iceberg-resource

export RESOURCE_SA=$(bq show --connection --project_id="$PROJECT_ID" \
  --location="$LOCATION" --format=json bq-iceberg-resource \
  | jq -r '.cloudResource.serviceAccountId')

gcloud storage buckets add-iam-policy-binding "gs://$GCS_BUCKET" \
  --member="serviceAccount:$RESOURCE_SA" \
  --role="roles/storage.objectAdmin"

CREATE OR REPLACE TABLE
  iceberg_dataset_tokyo.payment_log_bq_iceberg
WITH CONNECTION `[PROJECT_ID].asia-northeast1.bq-iceberg-resource`
OPTIONS (
  table_format = 'ICEBERG',
  storage_uri = 'gs://sb-iceberg-verify-mikami/bq_iceberg/'
)
AS SELECT * FROM
  iceberg_dataset_tokyo.payment_log_std;

# 1. Native テーブル
bq query --use_legacy_sql=false --nouse_cache \
"SELECT store_code, payment_method, SUM(amount) AS total, COUNT(*) AS cnt
 FROM \`[PROJECT_ID].iceberg_dataset_tokyo.payment_log_std\`
 GROUP BY store_code, payment_method ORDER BY total DESC"

# 2. 外部テーブル（Parquet）
bq query --use_legacy_sql=false --nouse_cache \
"SELECT store_code, payment_method, SUM(amount) AS total, COUNT(*) AS cnt
 FROM \`[PROJECT_ID].iceberg_dataset_tokyo.payment_log_ext\`
 GROUP BY store_code, payment_method ORDER BY total DESC"

# 3. Lakehouse カタログ Iceberg テーブル
bq query --use_legacy_sql=false --nouse_cache \
"SELECT store_code, payment_method, SUM(amount) AS total, COUNT(*) AS cnt
 FROM \`[PROJECT_ID].sb-iceberg-verify-mikami.sb_verify.payment_log_perf\`
 GROUP BY store_code, payment_method ORDER BY total DESC"

# 4. BigQuery 管理 Iceberg テーブル
bq query --use_legacy_sql=false --nouse_cache \
"SELECT store_code, payment_method, SUM(amount) AS total, COUNT(*) AS cnt
FROM \`[PROJECT_ID].iceberg_dataset_tokyo.payment_log_bq_iceberg\`
GROUP BY store_code, payment_method ORDER BY total DESC"

SELECT
  CASE
    WHEN REGEXP_CONTAINS(query, r'payment_log_std')  THEN '1.Native テーブル'
    WHEN REGEXP_CONTAINS(query, r'payment_log_ext')  THEN '2.外部テーブル（Parquet）'
    WHEN REGEXP_CONTAINS(query, r'payment_log_perf') THEN '3.Lakehouse Iceberg テーブル'
    WHEN REGEXP_CONTAINS(query, r'payment_log_bq_iceberg') THEN '4.BigQuery 管理 Iceberg テーブル'
    ELSE 'その他'
  END                                                            AS target_table,
  ROUND(AVG(TIMESTAMP_DIFF(end_time, start_time, MILLISECOND)) / 1000.0, 2)
                                                                 AS avg_elapsed_sec,
  ROUND(MIN(TIMESTAMP_DIFF(end_time, start_time, MILLISECOND)) / 1000.0, 2)
                                                                 AS min_elapsed_sec,
  ROUND(MAX(TIMESTAMP_DIFF(end_time, start_time, MILLISECOND)) / 1000.0, 2)
                                                                 AS max_elapsed_sec,
  ROUND(AVG(total_bytes_processed) / POW(1024, 3), 3)           AS avg_processed_gb
FROM `[PROJECT_ID]`.`region-asia-northeast1`.INFORMATION_SCHEMA.JOBS
WHERE
  statement_type  = 'SELECT'
  AND cache_hit   = false
  AND creation_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)
  AND REGEXP_CONTAINS(query, r'payment_log_(std|perf|ext|bq_iceberg)')
GROUP BY target_table
ORDER BY target_table;