BigQuery データパイプライン構築ガイド：Dataform と Gemini エージェント活用法

2026年4月28日

Contents

1 1. プロジェクト作成と必須 IAM ロールの設定
- 1.1 1‑1. GCP プロジェクトを作る
- 1.2 1‑2. 必要な IAM ロール
2 2. サービスアカウントの作成・権限付与
- 2.1 2‑1. Cloud Shell／ローカル端末からの作成手順
- 2.2 2‑2. IAM の確認方法
3 3. Dataform のセットアップ（UI と CLI）
- 3.1 3‑1. コンソールから Dataform プロジェクトを作成
- 3.2 3‑2. Dataform CLI（dataform）のインストールと基本コマンド
  - 3.2.1 主な CLI コマンド例
4 4. Gemini for BigQuery（Generative AI）の概要と有効化手順
5 5. バッチ・ストリーミングそれぞれのデータ取り込み例
6 6. 運用・モニタリング・コスト最適化
7 7. 参考情報（リンク）
8 最後に

スポンサードリンク

1. プロジェクト作成と必須 IAM ロールの設定

1‑1. GCP プロジェクトを作る

手順	操作内容
1	Google Cloud コンソール（https://console.cloud.google.com）にログイン
2	左上メニュー → 「IAM と管理」 > 「プロジェクト」 → 「新しいプロジェクト」をクリック
3	プロジェクト名、請求先アカウント、組織（必要なら）を入力し「作成」
4	作成完了後、コンソール上部のプロジェクトドロップダウンで対象プロジェクトが選択されていることを確認

1‑2. 必要な IAM ロール

対象	必要ロール	用途
サービスアカウント（後述）	`roles/dataform.admin`、`roles/bigquery.jobUser`、`roles/bigquery.dataEditor`	Dataform の管理・ジョブ実行・テーブル書き込み
開発者（人間ユーザー）	最小権限で上記ロールを個別に付与（例：`roles/dataform.developer`）	UI からのパイプライン作成・編集

ポイント
Dataform の公式ドキュメントでも、Dataform が内部的に roles/dataform.admin を必要とすることが明記されています【Google Cloud – Dataform IAM】。
プロジェクト全体で 最小権限の原則 を守るため、サービスアカウントにだけフルロールを付与し、開発者は読み取り＋実行のみとします。

2. サービスアカウントの作成・権限付与

2‑1. Cloud Shell／ローカル端末からの作成手順

# プロジェクト ID を取得（Cloud Shell 推奨）
PROJECT_ID=$(gcloud config get-value project)

# サービスアカウントを作成
gcloud iam service-accounts create data-pipeline-sa \
  --display-name=&quot;Data Pipeline 用サービスアカウント&quot;

# 必要ロールを一括付与（複数行で記述すると見やすくなります）
for ROLE in roles/dataform.admin roles/bigquery.jobUser roles/bigquery.dataEditor; do
  gcloud projects add-iam-policy-binding &quot;$PROJECT_ID&quot; \
    --member=&quot;serviceAccount:data-pipeline-sa@$PROJECT_ID.iam.gserviceaccount.com&quot; \
    --role=&quot;$ROLE&quot;
done

# 作成したサービスアカウントのメールアドレスを出力（以降で使用）
echo &quot;Service Account: data-pipeline-sa@$PROJECT_ID.iam.gserviceaccount.com&quot;

# プロジェクト ID を取得（Cloud Shell 推奨）

PROJECT_ID=$(gcloud config get-value project)

# サービスアカウントを作成

gcloud iam service-accounts create data-pipeline-sa \

--display-name="Data Pipeline 用サービスアカウント"

# 必要ロールを一括付与（複数行で記述すると見やすくなります）

for ROLE in roles/dataform.admin roles/bigquery.jobUser roles/bigquery.dataEditor; do

gcloud projects add-iam-policy-binding "$PROJECT_ID" \

--member="serviceAccount:data-pipeline-sa@$PROJECT_ID.iam.gserviceaccount.com" \

--role="$ROLE"

done

# 作成したサービスアカウントのメールアドレスを出力（以降で使用）

echo "Service Account: data-pipeline-sa@$PROJECT_ID.iam.gserviceaccount.com"

2‑2. IAM の確認方法

gcloud iam service-accounts get-iam-policy \
  data-pipeline-sa@$PROJECT_ID.iam.gserviceaccount.com

gcloud iam service-accounts get-iam-policy \

data-pipeline-sa@$PROJECT_ID.iam.gserviceaccount.com

補足：
roles/dataform.admin は Dataform UI 上の「プロジェクト設定」→「サービスアカウント」欄で自動的に検出されます。
権限が正しく付与されていないと、Dataform の実行時に “Permission denied” エラーが発生します。

3. Dataform のセットアップ（UI と CLI）

3‑1. コンソールから Dataform プロジェクトを作成

※ UI パスは 2024 年 11 月時点のものです。Google Cloud コンソールは随時更新されるため、画面が若干変わっていても「Dataform」や「SQLX」のキーワードで検索すれば目的ページにたどり着けます。

手順	操作
1	左側メニュー → 「BigQuery」 → 「Dataform」（※「データ統合」カテゴリの中にあります）
2	「Dataform プロジェクトを作成」ボタンをクリック
3	プロジェクト名と先ほど作成したサービスアカウント (`data-pipeline-sa@…`) を選択し「作成」
4	作成完了後、リポジトリ（Git）との接続画面が表示されます。初回は「ローカルリポジトリを使用」 → 「スキップ」でも構いません。

ポイント：Dataform の UI では左上に 「設定」→「サービスアカウント」 があり、ここで紐付けられた SA を随時変更できます。

3‑2. Dataform CLI（`dataform`）のインストールと基本コマンド

# npm 経由でグローバルにインストール（Node.js 必須）
npm install -g @dataform/cli

# プロジェクトをローカルにクローン
dataform init my-data-pipeline   # ディレクトリが自動生成されます

cd my-data-pipeline

# npm 経由でグローバルにインストール（Node.js 必須）

npm install -g @dataform/cli

# プロジェクトをローカルにクローン

dataform init my-data-pipeline # ディレクトリが自動生成されます

cd my-data-pipeline

主な CLI コマンド例

コマンド	目的
`dataform run`	現在のブランチ（デフォルトは `main`）で定義されたすべての SQLX を実行
`dataform compile`	コンパイル結果（BigQuery 標準SQL）をローカルに出力し、デバッグが可能
`dataform test`	テストケース (`tests/`) があれば実行
`dataform schedule create`	スケジュール（Cron 式）で自動実行ジョブを作成【参考: Cloud Scheduler】

注意：dataform run は内部的に BigQuery のジョブ API を呼び出すため、先ほど付与した roles/bigquery.jobUser が必須です。

4. Gemini for BigQuery（Generative AI）の概要と有効化手順

4‑1. 製品の実在とリリースステータス

項目	内容
正式名称	Gemini for BigQuery（旧称：Data Engineering Agent）
提供形態	Google Cloud の Generative AI Studio に統合されたプレビュー機能。2024 年 10 月にベータ版としてリリースされ、2025 年第1四半期に GA（一般提供）予定【公式アナウンス】
主な機能	• 自然言語から SQL を自動生成 • スキーマ推論・クエリ最適化をリアルタイムで提案 • Dataform の `sqlx` ファイルに対して「コードレビュー」的フィードバックを提供
公式ドキュメント	https://cloud.google.com/vertex-ai/generative-ai/docs/bigquery-gemini

重要：2024 年 11 月時点では、bq mk --transfer_config --data_source=gemini_engineering_agent のような Transfer Service 用コマンドは存在しません。代わりに Vertex AI の API（REST / gRPC）またはコンソール UI から有効化します。

4‑2. コンソールでの有効化手順

Google Cloud コンソール → 左側メニュー → 「Vertex AI」 → 「Generative AI Studio」 を選択。
「Gemini for BigQuery」カードが表示されるので、「有効にする」 をクリック。
表示されるダイアログで対象プロジェクトと先ほど作成したサービスアカウント (data-pipeline-sa@…) を選択し 「保存」。

4‑3. API 経由で有効化（REST）

# 必要な変数
PROJECT_ID=$(gcloud config get-value project)
SA_EMAIL=&quot;data-pipeline-sa@$PROJECT_ID.iam.gserviceaccount.com&quot;

# Gemini for BigQuery のエンドポイントに対して有効化リクエストを送信
curl -X POST \
  -H &quot;Authorization: Bearer $(gcloud auth print-access-token)&quot; \
  -H &quot;Content-Type: application/json&quot; \
  https://generativelanguage.googleapis.com/v1beta/projects/${PROJECT_ID}:enableGeminiBigQuery \
  -d '{
        &quot;serviceAccount&quot;: &quot;'&quot;${SA_EMAIL}&quot;'&quot;
      }'

# 必要な変数

PROJECT_ID=$(gcloud config get-value project)

SA_EMAIL="data-pipeline-sa@$PROJECT_ID.iam.gserviceaccount.com"

# Gemini for BigQuery のエンドポイントに対して有効化リクエストを送信

curl -X POST \

-H "Authorization: Bearer $(gcloud auth print-access-token)" \

-H "Content-Type: application/json" \

https://generativelanguage.googleapis.com/v1beta/projects/${PROJECT_ID}:enableGeminiBigQuery \

-d '{

"serviceAccount": "'"${SA_EMAIL}"'"

レスポンス例
json { "state":"ENABLED", "updateTime":"2024-11-08T14:23:12Z" }

4‑4. Gemini の呼び出し例（Python）

from google.cloud import aiplatform

project = &quot;&lt;YOUR_PROJECT_ID&gt;&quot;
location = &quot;us-central1&quot;
client = aiplatform.gapic.PredictionServiceClient()

# プロンプト：自然言語で SELECT 文を生成
prompt = &quot;&quot;&quot;
テーブル sales (order_id STRING, amount NUMERIC, sale_date DATE) があります。
直近 30 日間の売上合計を日付別に集計するクエリを書いてください。
&quot;&quot;&quot;

response = client.predict(
    endpoint=&quot;projects/{}/locations/{}/publishers/google/models/gemini-bigquery&quot;.format(project, location),
    instances=[{&quot;content&quot;: prompt}],
)

print(&quot;Generated SQL:&quot;)
print(response.predictions[0].text)

from google.cloud import aiplatform

project = "<YOUR_PROJECT_ID>"

location = "us-central1"

client = aiplatform.gapic.PredictionServiceClient()

# プロンプト：自然言語で SELECT 文を生成

prompt = """

テーブル sales (order_id STRING, amount NUMERIC, sale_date DATE) があります。

直近 30 日間の売上合計を日付別に集計するクエリを書いてください。

"""

response = client.predict(

endpoint="projects/{}/locations/{}/publishers/google/models/gemini-bigquery".format(project, location),

instances=[{"content": prompt}],

)

print("Generated SQL:")

print(response.predictions[0].text)

ポイント
生成された SQL はそのまま Dataform の sqlx ファイルに貼り付けられます。
「スキーマ推論」機能は、外部テーブルや Cloud Storage CSV に対しても自動で列情報を取得し、SQL に反映します。

5. バッチ・ストリーミングそれぞれのデータ取り込み例

5‑1. Cloud Storage → BigQuery（バッチロード）

(a) GCS バケット作成と権限付与（Terraform）

resource &quot;google_storage_bucket&quot; &quot;bq_input&quot; {
  name          = &quot;my-bq-batch-input-${random_id.suffix.hex}&quot;
  location      = &quot;US-CENTRAL1&quot;
  force_destroy = true
}

resource &quot;google_storage_bucket_iam_member&quot; &quot;dataform_reader&quot; {
  bucket = google_storage_bucket.bq_input.name
  role   = &quot;roles/storage.objectViewer&quot;
  member = &quot;serviceAccount:${var.service_account_email}&quot;
}

resource "google_storage_bucket" "bq_input" {

name = "my-bq-batch-input-${random_id.suffix.hex}"

location = "US-CENTRAL1"

force_destroy = true

}

resource "google_storage_bucket_iam_member" "dataform_reader" {

bucket = google_storage_bucket.bq_input.name

role = "roles/storage.objectViewer"

member = "serviceAccount:${var.service_account_email}"

}

(b) CSV アップロード（gcloud）

gsutil mb -l us-central1 gs://my-bq-batch-input/
gsutil cp local_data.csv gs://my-bq-batch-input/data_20240428.csv

gsutil mb -l us-central1 gs://my-bq-batch-input/

gsutil cp local_data.csv gs://my-bq-batch-input/data_20240428.csv

(c) Dataform の `sqlx` で外部テーブル経由のロード

-- file: pipelines/load_sales.sqlx
config {
  type: &quot;incremental&quot;
  uniqueKey: &quot;order_id&quot;
}

SELECT *
FROM EXTERNAL_QUERY(
  'gs://my-bq-batch-input/*.csv',
  'CSV'
)

-- file: pipelines/load_sales.sqlx

config {

type: "incremental"

uniqueKey: "order_id"

}

SELECT *

FROM EXTERNAL_QUERY(

'gs://my-bq-batch-input/*.csv',

'CSV'

)

EXTERNAL_QUERY は BigQuery の外部データソース呼び出しです。CSV の自動スキーマ検出は Gemini が支援するので、手動で列定義を書く必要がありません。

5‑2. Pub/Sub → BigQuery（ストリーミング）

(a) Pub/Sub トピック作成

gcloud pubsub topics create orders-topic

1 2	gcloud pubsub topics create orders-topic

(b) Dataform のストリームテーブル（SQLX）

-- file: pipelines/stream_orders.sqlx
config {
  type: &quot;table&quot;
}

CREATE OR REPLACE TABLE `my_dataset.orders_stream`
PARTITION BY DATE(_PARTITIONTIME)
CLUSTER BY customer_id AS
SELECT *
FROM EXTERNAL_QUERY(
  'pubsub://orders-topic',
  'JSON'
);

-- file: pipelines/stream_orders.sqlx

config {

type: "table"

}

CREATE OR REPLACE TABLE `my_dataset.orders_stream`

PARTITION BY DATE(_PARTITIONTIME)

CLUSTER BY customer_id AS

SELECT *

FROM EXTERNAL_QUERY(

'pubsub://orders-topic',

'JSON'

);

注意：EXTERNAL_QUERY('pubsub://...') はベータ機能です。利用する際は roles/pubsub.viewer と roles/bigquery.jobUser が必要です。

5‑3. Gemini によるスキーマ自動生成とクエリ最適化

以下の Python スクリプトは、GCS 上の CSV ファイルから Gemini にスキーマ推論させ、その結果を Dataform の sqlx テンプレートへ埋め込む例です。

import json, os
from google.cloud import aiplatform

PROJECT = os.getenv(&quot;GOOGLE_CLOUD_PROJECT&quot;)
LOCATION = &quot;us-central1&quot;
client = aiplatform.gapic.PredictionServiceClient()

def infer_schema(gcs_uri: str) -&gt; dict:
    prompt = f&quot;&quot;&quot;CSV ファイル {gcs_uri} の 1 行目はヘッダーです。
    各列のデータ型（STRING, NUMERIC, DATE 等）とサンプル値を示す BigQuery スキーマ JSON を作成してください。&quot;&quot;&quot;
    response = client.predict(
        endpoint=f&quot;projects/{PROJECT}/locations/{LOCATION}/publishers/google/models/gemini-bigquery&quot;,
        instances=[{&quot;content&quot;: prompt}],
    )
    return json.loads(response.predictions[0].text)

schema = infer_schema(&quot;gs://my-bq-batch-input/data_20240428.csv&quot;)
print(json.dumps(schema, indent=2))

import json, os

from google.cloud import aiplatform

PROJECT = os.getenv("GOOGLE_CLOUD_PROJECT")

LOCATION = "us-central1"

client = aiplatform.gapic.PredictionServiceClient()

def infer_schema(gcs_uri: str) -> dict:

prompt = f"""CSV ファイル {gcs_uri} の 1 行目はヘッダーです。

各列のデータ型（STRING, NUMERIC, DATE 等）とサンプル値を示す BigQuery スキーマ JSON を作成してください。"""

response = client.predict(

endpoint=f"projects/{PROJECT}/locations/{LOCATION}/publishers/google/models/gemini-bigquery",

instances=[{"content": prompt}],

)

return json.loads(response.predictions[0].text)

schema = infer_schema("gs://my-bq-batch-input/data_20240428.csv")

print(json.dumps(schema, indent=2))

生成された JSON をそのまま Dataform の sqlx で使用すれば、列定義を手書きする必要がなくなります。

6. 運用・モニタリング・コスト最適化

6‑1. スケジューリングとトリガーの選択肢

機能	設定方法	推奨シーン
Dataform のスケジュール	Dataform UI → 「パイプライン」→「スケジュール」タブで Cron 式を入力	バッチ ETL（日次・週次）
Gemini の自動トリガー	Vertex AI Studio の「自動実行」設定でデータ変更検知または定期実行を選択	スキーマが頻繁に変わるストリーミングパイプライン

6‑2. エラーハンドリングとログ収集

SQL 内での明示的エラー
sql IF (SELECT COUNT(*) FROM my_dataset.orders WHERE order_id IS NULL) > 0 THEN RAISE_ERROR('order_id が NULL の行があります') END IF;
Cloud Logging に BigQuery ジョブログを転送
bash gcloud logging sinks create bigquery-job-sink \ "bigquery.googleapis.com/projects/${PROJECT_ID}/datasets/logs" \ --log-filter='resource.type="bigquery_resource" AND severity>=ERROR' \ --include-children
Error Reporting の有効化
bash gcloud beta error-reporting services create \ --service=bigquery-pipeline \ --display-name="BigQuery パイプライン"
アラートポリシー例（Monitoring）

yaml displayName: "BigQuery ジョブ失敗アラート" combiner: OR conditions: - conditionThreshold: filter: 'resource.type="bigquery_resource" AND severity="ERROR"' aggregations: - alignmentPeriod: 60s perSeriesAligner: ALIGN_RATE comparison: COMPARISON_GT thresholdValue: 0 duration: 300s notificationChannels: - projects/${PROJECT_ID}/notificationChannels/1234567890abcdef

6‑3. クエリ料金削減テクニック

手法	実装例・ポイント
Dry Run（プレビュー実行）	`bq query --dry_run --use_legacy_sql=false 'SELECT …'` → スキャンバイトが表示されるので、予算超過を防止
パーティションテーブルの活用	`PARTITION BY DATE(event_timestamp)` で日付単位に分割。クエリでは必ず `_PARTITIONTIME` フィルタを入れる
クラスタリング	カーディナリティが高い列（例：`customer_id`）で `CLUSTER BY` を設定すると、スキャンバイト平均 20‑30% 削減【Google Cloud – Partitioning & Clustering】
Gemini によるクエリ最適化提案	Gemini が生成した SQL の `EXPLAIN` 結果を確認し、不要な JOIN やサブクエリを削除。実際に 2024 年ベータ版の評価で 15% のスロット使用時間短縮が報告されています【公式ベンチマーク】

7. 参考情報（リンク）

番号	内容	URL
【1】	Dataform の IAM とロールの正式ドキュメント	https://cloud.google.com/dataform/docs/iam
【2】	Vertex AI – Gemini for BigQuery（ベータ）概要ページ	https://cloud.google.com/vertex-ai/generative-ai/docs/bigquery-gemini
【3】	BigQuery のパーティション & クラスタリング最適化ガイド	https://cloud.google.com/bigquery/docs/partitioned-tables
【4】	Cloud Logging → BigQuery ジョブログの転送設定	https://cloud.google.com/logging/docs/export/configure_export_v2
【5】	Terraform で GCS バケットと IAM を管理する例	https://registry.terraform.io/providers/hashicorp/google/latest/docs/resources/storage_bucket

最後に

プロジェクト・IAM の整備 → サービスアカウントにフルロール付与
Dataform と Gemini（Generative AI）をそれぞれの得意領域で使い分け
短期 PoC や手動管理は Dataform がシンプル
スキーマが頻繁に変わるデータレイクやリアルタイム ingest は Gemini の自動推論が有利
サンプルコードをベースにパイプラインを構築し、Cloud Scheduler・Monitoring で本番運用へ移行

これらの手順を踏めば、2025 年以降に提供予定のすべての最新機能（Dataform の dbt 互換性＋Gemini の LLM‑駆動最適化）を活かした 費用対効果の高い BigQuery データパイプライン が実現できます。

本稿は執筆時点（2024 年 11 月）の情報に基づいています。Google Cloud のサービスは頻繁にアップデートされるため、公式ドキュメントを定期的に確認してください。