スクリプト・UDF・ELT パイプライン

スクリプトで手続き型ロジックを書く

これまでの章で書いてきた SQL は「1 つの SELECT / DDL / DML 文」が単位でした。GoogleSQL の手続き型言語（procedural language）を使うと、複数の文をセミコロンで連結したマルチステートメントクエリとして、1 つのジョブの中で順番に実行できます。変数を宣言して状態を持ち回したり、条件分岐やループでロジックを組んだりできるため、「テーブルが存在すれば洗い替え、なければ新規作成する」「一定件数ずつ処理してリトライする」といった、単純な SELECT では書けない自動化が可能になります。

DECLARE で変数を宣言し、SET で値を更新します。変数の型は明示するか、DEFAULT 句の式から推論されます。宣言はブロックの先頭にまとめて書く必要があります。

DECLARE target_word STRING DEFAULT 'methinks';
DECLARE corpus_count INT64;
DECLARE word_count INT64;

-- 処理: 数 MB 未満（shakespeare は小さい公開データセット）
SET (corpus_count, word_count) = (
  SELECT AS STRUCT COUNT(DISTINCT corpus), SUM(word_count)
  FROM `bigquery-public-data.samples.shakespeare`
  WHERE LOWER(word) = target_word
);

IF word_count > 100 THEN
  SELECT FORMAT('%s は %d 回登場（頻出語）', target_word, word_count) AS result;
ELSE
  SELECT FORMAT('%s は %d 回登場（希少語）', target_word, word_count) AS result;
END IF;

繰り返し処理には LOOP / WHILE / REPEAT / FOR...IN があります。FOR record IN (SELECT ...) DO ... END FOR はクエリ結果を 1 行ずつ変数に受けて処理する形で、他言語のカーソルに近い書き方です。ループを抜けるには BREAK（別名 LEAVE）、次の反復に進むには CONTINUE（別名 ITERATE）を使います。ブロックや条件分岐のネストは最大 50 段までという制約があります。

エラー処理は BEGIN...EXCEPTION WHEN ERROR THEN...END で行います。BEGIN 節でエラーが起きると、残りの文はスキップされ EXCEPTION 節に処理が移ります。EXCEPTION 節の中では @@error.message（人間可読なエラーメッセージ）・@@error.stack_trace（呼び出し元をたどる配列）・@@error.statement_text（エラーを起こした文のテキスト）という 3 つのシステム変数でエラーの詳細を取得できます。これらは EXCEPTION 節の外では参照できません。

BEGIN
  DECLARE attempt INT64 DEFAULT 0;
  DECLARE max_attempts INT64 DEFAULT 3;

  LOOP
    SET attempt = attempt + 1;
    IF attempt > max_attempts THEN
      LEAVE;
    END IF;
    -- ここに実際の処理（例: 外部呼び出しのリトライ対象）を書く
  END LOOP;

  SELECT FORMAT('%d 回試行しました', attempt) AS result;
EXCEPTION WHEN ERROR THEN
  SELECT
    @@error.message AS error_message,
    @@error.statement_text AS failed_statement;
END;

動的に組み立てた SQL 文字列を実行したいときは EXECUTE IMMEDIATE を使います。? や @identifier プレースホルダで USING 句から値を渡し、INTO 句で結果を変数へ受け取れます。ただし EXECUTE IMMEDIATE の中に IF や BEGIN などの制御構文をネストすることはできません（動的実行できるのは単一の SELECT / DDL / DML / DCL 文のみ）。

UDF とリモート関数でロジックを部品化する

ユーザー定義関数（UDF）は、繰り返し使う計算やロジックを名前付きの関数として部品化する仕組みです。CREATE FUNCTION で作った永続 UDF はデータセットに保存され、プロジェクトをまたいで複数のクエリ・複数の利用者から呼び出せます。同じデータセットに同名の永続 UDF は 1 つしか置けません。一方 CREATE TEMP FUNCTION で作る一時 UDF はクエリの実行中だけ有効で、クエリが終わると同時に消えます。使い捨ての変換ロジックや実験段階の式は一時 UDF、チームで共有する定義は永続 UDF、と使い分けます。

CREATE OR REPLACE FUNCTION handson_ch11.price_with_tax(price NUMERIC, tax_rate NUMERIC)
RETURNS NUMERIC AS (
  ROUND(price * (1 + tax_rate), 0)
);

SELECT
  product_name,
  price,
  handson_ch11.price_with_tax(price, 0.10) AS price_with_tax
FROM UNNEST([
  STRUCT('ノート PC' AS product_name, 98000 AS price),
  STRUCT('マウス' AS product_name, 2400 AS price)
]);

SQL で表現できるロジックは SQL UDF として書くのが基本です。SQL では書きにくい文字列処理や正規表現を伴う変換には JavaScript UDF が使えますが、公式ドキュメントは「JavaScript UDF は標準的な SQL クエリより多くのスロットを消費し、ジョブのパフォーマンスを下げる傾向がある」としています。また JavaScript の数値表現の制約により、JavaScript UDF の引数に INT64 を直接渡すことはできません（FLOAT64 か STRING にキャストします）。外部の JavaScript ライブラリは OPTIONS (library=['gs://bucket/lib.js']) で Cloud Storage 上のファイルを読み込ませます。

CREATE OR REPLACE FUNCTION handson_ch11.slugify(input STRING)
RETURNS STRING
LANGUAGE js AS r"""
  return input.toLowerCase()
    .replace(/[^a-z0-9]+/g, '-')
    .replace(/(^-|-$)/g, '');
""";

SELECT handson_ch11.slugify('BigQuery スクリプト & UDF!') AS slug;

永続 UDF を安全に共有する仕組みが認可ルーチン（authorized routines）です。IAM で UDF 自体への実行権限を付与しつつ、UDF が参照する基盤テーブルへの直接アクセスは与えないという構成にできるため、「集計済みの結果だけを見せたい」という要件に使えます。

UDF が「BigQuery の中で完結するロジック」だとすれば、リモート関数は「BigQuery の外にあるロジックをクエリから呼び出す」仕組みです。SQL の CREATE FUNCTION ... REMOTE WITH CONNECTION で、Cloud Run functions（または Cloud Run）の HTTP エンドポイントを 1 つの関数として登録します。クエリ実行時、BigQuery は複数行の引数をバッチにまとめて JSON の POST リクエストを送り、エンドポイントは同じ順序で結果を返します。

CREATE FUNCTION handson_ch11.classify_sentiment(review_text STRING)
RETURNS STRING
REMOTE WITH CONNECTION `PROJECT_ID.us.remote-conn`
OPTIONS (
  endpoint = 'https://REGION-PROJECT_ID.cloudfunctions.net/classifySentiment'
);

接続には CLOUD_RESOURCE タイプの接続（connection）が必須です。接続作成時に自動生成されるサービスアカウントへ、呼び出し先の Cloud Run functions / Cloud Run を起動できる IAM ロール（invoker）を付与しておく必要があります。エンドポイント側が受け取るリクエストと返すレスポンスの形は固定の契約です。

// リクエスト（BigQuery → エンドポイント）
{
  "requestId": "124ab1c",
  "caller": "//bigquery.googleapis.com/projects/PROJECT_ID/jobs/JOB_ID",
  "calls": [["とても良い商品でした"], ["普通です"]]
}

// レスポンス（エンドポイント → BigQuery）
{"replies": ["positive", "neutral"]}

リモート関数の引数・戻り値は BOOL / BYTES / NUMERIC / STRING / DATE / DATETIME / TIME / TIMESTAMP / JSON のいずれか 1 つに限られ、ARRAY・STRUCT・INTERVAL・GEOGRAPHY はサポートされません。複合的な結果を返したい場合は JSON 型にシリアライズします。結果はキャッシュされず（非決定論的とみなされるため）、マテリアライズドビューの定義にも使えません。max_batching_rows オプションでバッチの最大行数を絞り、エンドポイント側のタイムアウトを避けられます。料金は通常の BigQuery 課金に加えて、呼び出し先の Cloud Run / Cloud Run functions の実行料金が発生します。

ストアドプロシージャ・テーブル関数と ELT の設計思想

ストアドプロシージャは CREATE PROCEDURE で定義し、BEGIN...END の中にスクリプトと同じ制御構文（IF・LOOP・EXCEPTION）を書けます。引数には修飾子なしの入力専用パラメータのほか、値を返す OUT、入出力を兼ねる INOUT が使えます。呼び出しは CALL 文です。

CREATE OR REPLACE PROCEDURE handson_ch11.refresh_mart(OUT rows_affected INT64)
BEGIN
  CREATE OR REPLACE TABLE handson_ch11.mart_word_by_corpus AS
  SELECT corpus, COUNT(DISTINCT word) AS distinct_words, SUM(word_count) AS total_words
  FROM handson_ch11.stg_word_counts
  GROUP BY corpus;

  SET rows_affected = (SELECT COUNT(*) FROM handson_ch11.mart_word_by_corpus);
END;

DECLARE affected INT64;
CALL handson_ch11.refresh_mart(affected);
SELECT affected AS mart_rows;

ストアドプロシージャと似て非なるものがテーブル関数（TVF）です。CREATE TABLE FUNCTION で定義し、ビューのようにテーブルを返しますが、ビューと違ってパラメータを取れます。本体は SELECT 文 1 本に限られ、DDL や DML、制御構文は書けません。副作用（テーブル作成・更新）が必要な処理はプロシージャ、パラメータ付きの読み取り専用ビューが欲しいときは TVF、と役割が分かれます。

CREATE OR REPLACE TABLE FUNCTION handson_ch11.words_by_corpus(target_corpus STRING)
AS (
  SELECT word, word_count
  FROM `bigquery-public-data.samples.shakespeare`
  WHERE corpus = target_corpus
  ORDER BY word_count DESC
  LIMIT 20
);

SELECT * FROM handson_ch11.words_by_corpus('hamlet');

ここまでの部品（スクリプト・UDF・プロシージャ）は、ELT パイプラインという 1 つの設計に組み上がっていきます。本書で扱う最小の設計原型は、raw → staging → mart という 3 層のデータセット構造です。raw は取り込んだ生データをそのまま保持し、後から再処理できる状態を残します。staging では型の整備や重複排除、名寄せといったクレンジングを行います。mart は集計済みでレポーティングや BI ツールから直接参照される層です。層を分ける理由は、障害が起きたときにどの層まで戻ればよいかを切り分けやすくすること、そして staging より上流を作り直す必要が出ても mart のスキーマを変えずに済ませられることにあります。第 9 章で扱った DML（特に MERGE による upsert）は staging → mart の更新でよく使われ、この最小構成の完成形は第 18 章の総合演習で組み上げます。

ハンズオン

DECLARE attempt INT64 DEFAULT 0;
DECLARE max_attempts INT64 DEFAULT 3;
DECLARE target_word STRING DEFAULT 'sword';
DECLARE word_count INT64;

BEGIN
  LOOP
    SET attempt = attempt + 1;

    SET word_count = (
      SELECT SUM(word_count)
      FROM `bigquery-public-data.samples.shakespeare`
      WHERE LOWER(word) = target_word
    );

    IF word_count IS NOT NULL OR attempt >= max_attempts THEN
      LEAVE;
    END IF;
  END LOOP;

  SELECT
    target_word,
    word_count,
    attempt AS attempts_used;
EXCEPTION WHEN ERROR THEN
  SELECT @@error.message AS error_message;
END;

CREATE SCHEMA IF NOT EXISTS handson_ch11
OPTIONS (location = 'US');

CREATE OR REPLACE FUNCTION handson_ch11.price_with_tax(price NUMERIC, tax_rate NUMERIC)
RETURNS NUMERIC AS (
  ROUND(price * (1 + tax_rate), 0)
);

SELECT
  product_name,
  price,
  handson_ch11.price_with_tax(price, 0.10) AS price_with_tax
FROM UNNEST([
  STRUCT('ノート PC' AS product_name, 98000 AS price),
  STRUCT('マウス' AS product_name, 2400 AS price)
]);

DROP FUNCTION IF EXISTS handson_ch11.price_with_tax;

-- 処理: 6 MB 未満（shakespeare は小さい公開データセット、corpus を 3 作品に絞る）
CREATE OR REPLACE TABLE handson_ch11.raw_shakespeare AS
SELECT word, word_count, corpus, corpus_date
FROM `bigquery-public-data.samples.shakespeare`
WHERE corpus IN ('hamlet', 'macbeth', 'othello');

CREATE OR REPLACE TABLE handson_ch11.stg_word_counts AS
SELECT
  LOWER(word) AS word,
  corpus,
  word_count,
  corpus_date
FROM handson_ch11.raw_shakespeare
WHERE word_count > 0;

CREATE OR REPLACE TABLE handson_ch11.mart_word_by_corpus AS
SELECT
  corpus,
  COUNT(DISTINCT word) AS distinct_words,
  SUM(word_count) AS total_words
FROM handson_ch11.stg_word_counts
GROUP BY corpus;

bq query \
  --use_legacy_sql=false \
  --display_name="ch11_daily_mart_refresh" \
  --schedule="every 24 hours" \
  --destination_table=PROJECT_ID:handson_ch11.mart_word_by_corpus \
  --replace=true \
'SELECT corpus, COUNT(DISTINCT word) AS distinct_words, SUM(word_count) AS total_words
 FROM `PROJECT_ID.handson_ch11.stg_word_counts`
 GROUP BY corpus'

-- スケジュールクエリの削除は BigQuery Studio の Scheduled queries 画面から対象を選び Delete
-- テーブル・データセットは章末でまとめて削除（次のハンズオンでも使うため、ここでは削除しない）

defaultProject: PROJECT_ID
defaultDataset: handson_ch11
dataformCoreVersion: CORE_VERSION

config { type: "table" }

SELECT word, word_count, corpus, corpus_date
FROM `bigquery-public-data.samples.shakespeare`
WHERE corpus IN ('hamlet', 'macbeth', 'othello')

config { type: "table" }

SELECT
  LOWER(word) AS word,
  corpus,
  word_count
FROM ${ref("raw_shakespeare")}
WHERE word_count > 0

config {
  type: "table",
  assertions: { uniqueKey: ["corpus"] }
}

SELECT
  corpus,
  COUNT(DISTINCT word) AS distinct_words,
  SUM(word_count) AS total_words
FROM ${ref("stg_word_counts")}
GROUP BY corpus

bq rm -r -f -d PROJECT_ID:handson_ch11

つまずきポイント

理解度チェック

まとめと次章へ

• スクリプト（DECLARE / SET / IF / LOOP + BEGIN...EXCEPTION...END）で、SQL に手続き型のロジックとエラー処理を組み込める
• UDF は SQL を優先し、JavaScript UDF・リモート関数は「SQL で書けない」場合の選択肢として使う
• ストアドプロシージャは複雑な処理と OUT パラメータに、テーブル関数はパラメータ付きの読み取り専用ビューに向く
• raw → staging → mart の 3 層構造を、依存関係が単純ならスケジュールクエリ、複雑ならアサーション付きの Dataform で自動化する
• 第 9 章の DML（特に MERGE）は staging → mart の更新でよく使われ、この最小構成は第 18 章の総合演習で完成形に組み上げる

参考リンク:

• Procedural language（GoogleSQL 手続き型言語リファレンス）
• User-defined functions
• Remote functions
• Stored procedures
• Table functions
• Scheduling queries
• BigQuery sandbox
• Dataform overview
• Dataform quickstart: Create and run a workflow