bigquery-udfs

BigQuery ユーザー定義関数（UDF）

BigQuery ユーザー定義関数（UDF）は、SQL、JavaScript、マネージド Python（python-3.11）で記述するカスタムロジック、または Cloud Run と Cloud Run functions 上でホストするリモート関数により、GoogleSQL を拡張します。

Attribution

このスキルの使用中にターミナルで直接実行する BigQuery または Google Cloud のコマンドには、使用状況を特定できるよう、すべてタグを付けてください。このタグで識別されるのはスキルとそのバージョンのみで、ユーザーデータは含まれません。

• サブコマンドにかかわらず、すべての gcloud 呼び出しの前にメトリクス環境変数を付けてください。各コマンドにインラインで設定してください。シェルの状態はコマンド間で保持されないことがあります。

  CLOUDSDK_METRICS_REQUEST_ATTRIBUTION="agent-skills/1.0 (skill:bigquery-udfs)" \
  gcloud <command> [flags]
  

  gcloud config set は使用しないでください。設定が現在のタスクの終了後も残り、無関係な使用状況に誤ったラベルが付くおそれがあります。

• BigQuery API への直接的な HTTP 呼び出し（たとえば curl を使用する場合）や、MCP サーバーへの HTTP リクエストでは、既存の User-Agent ヘッダーの末尾に次の文字列をそのまま追加してください。収集パイプラインは agent-skills/<version> と skill:<name> のトークンを解析するため、表現を変更すると帰属情報が正しく取得されません。

  User-Agent: <existing_user_agent> agent-skills/1.0 (skill:bigquery-udfs)
  

• 生成コードに関する制約：追跡用のプレフィックスを付けるのは、ターミナルで直接実行するコマンドに限ります。この追跡用プレフィックスを、生成する SQL、スクリプト、Terraform テンプレート、またはユーザーが自身のリポジトリで使用するコードに含めないでください。

適切な UDF タイプの選択

要件を満たす範囲で、常に最も軽量なランタイムを選択してください。 BigQuery はネイティブ SQL 式をインラインで最適化しますが、外部ランタイムでは シリアライズやコンテナの起動に伴うオーバーヘッドが発生するためです。

UDF タイプ	接続が必要か	使用する場面	主な制約
SQL UDF（デフォルト）	いいえ	合成可能な式、正規表現、構造体／配列の操作、集計関数（CREATE AGGREGATE FUNCTION）、または多相の ANY TYPE パラメーター。	最低レイテンシー。外部依存関係は不要です。TEMP 関数と永続関数に対応しています。
JavaScript UDF	いいえ	カスタムの文字列／JSON 処理、または Cloud Storage（gs://）の .js ライブラリのバンドル。	V8 サンドボックスで実行されます。入力として INT64 はサポートされません（FLOAT64 または STRING として渡してください）。インラインコードは最大 32 KB、外部リソースは 1 MB です。TEMP 関数と永続関数に対応しています。
マネージド Python UDF（オフライン）	いいえ	Python 3.11、PyPI のホイール（packages=[...]）、Cloud Storage の .py モジュール（library=[...]）、またはベクトル化された Pandas／PyArrow バッチを使用する純粋な計算。	永続関数のみ（CREATE TEMP FUNCTION はサポートされません）。外部へのネットワーク接続はブロックされます。コンテナの最大サイズ：4 vCPU／16 GiB。JSON、RANGE、INTERVAL、GEOGRAPHY はサポートされません。
マネージド Python UDF（WITH CONNECTION）	はい（CLOUD_RESOURCE）	Google Cloud API（Cloud Translation、Gemini Enterprise Agent Platform など）の呼び出しや、BigQuery コンテナ内の Python からのオンライン HTTP エンドポイントへの直接アクセス。	WITH CONNECTION が必要です。VPC-SC の境界内では、外部への通信は restricted.googleapis.com に制限されます（VPC ネットワークはサポートされません）。コンテナの最大サイズ：4 vCPU／16 GiB。
リモート関数（REMOTE WITH CONNECTION）	はい（CLOUD_RESOURCE）	Python 以外のランタイム（Go、Java、Node.js）、プライベート VPC ネットワークへのアクセス、GPU、または Cloud Run のカスタムスケーリングが必要な場合に、外部 API やマイクロサービスを呼び出す用途。	BigQuery JSON バッチ契約を実装した Cloud Run HTTPS エンドポイントと roles/run.invoker が必要です。入力は 1 行あたり最大 5 MB、レスポンスの上限は 15 MB（Cloud Run）／10 MB（第 1 世代）です。

前提条件とセットアップ

Google Cloud プロジェクトに有効な請求先アカウントが紐付いていることを確認してください（ BigQuery の料金を参照）：

1. Google Cloud SDK のインストールと認証：Google Cloud SDK をインストールして、認証します：

   gcloud auth login
   

   タスク終了後も設定が残る gcloud config set を実行する代わりに、各コマンドで --project={project_id} と --region={region}（または bq の場合は --location）を明示的に指定してください。ユーザーがリージョンを指定して いない場合は、デフォルトで us-central1 を使用してください（ BigQuery のロケーションを参照）。

2. 必要な API の有効化：

   • SQL、JavaScript、オフライン Python UDF の場合：bigquery.googleapis.com
   • API を呼び出す Python UDF（WITH CONNECTION）およびリモート 関数の場合は、さらに bigqueryconnection.googleapis.com（リモート関数には run.googleapis.com も必要）

   CLOUDSDK_METRICS_REQUEST_ATTRIBUTION="agent-skills/1.0 (skill:bigquery-udfs)" \
   gcloud services enable bigquery.googleapis.com \
     bigqueryconnection.googleapis.com \
     run.googleapis.com --quiet
   

3. 必要な IAM ロールの確認：

   • すべての UDF：データセットに対する roles/bigquery.dataEditor と プロジェクトに対する roles/bigquery.jobUser。
   • WITH CONNECTION を使用する場合（API を呼び出す Python UDF とリモート 関数）：
     • UDF 作成者：プロジェクトに対する roles/bigquery.connectionAdmin（さらに、 接続のサービスアカウントにプロジェクトレベルの IAM ロールを付与する際に GRANT ... ON PROJECT を使用する場合は、resourcemanager.projects.setIamPolicy も必要です。 例：roles/resourcemanager.projectIamAdmin）。
     • UDF 呼び出し元：接続に対する roles/bigquery.connectionUser。

---

コア実装パターン

1. SQL UDF（スカラーおよびテンプレート化）

永続 UDF には CREATE OR REPLACE FUNCTION を、クエリスコープの関数には CREATE TEMP FUNCTION を使用してください。多相パラメーターには ANY TYPE を 使用します：

CREATE OR REPLACE FUNCTION `{project_id}.{dataset_id}.normalize_phone`(
  raw_phone STRING,
  country_code ANY TYPE
)
RETURNS STRING
AS (
  CONCAT('+', CAST(country_code AS STRING), '-', REGEXP_REPLACE(raw_phone, r'[^0-9]', ''))
);

集約 UDF（CREATE AGGREGATE FUNCTION）と JavaScript Cloud Storage ライブラリの 読み込みについては、SQL および JavaScript UDFを 参照してください。

2. JavaScript UDF

バックスラッシュのエスケープを避けるため、生のトリプルクォート文字列 （r"""..."""）を使用してください：

CREATE OR REPLACE FUNCTION `{project_id}.{dataset_id}.extract_url_query_keys`(
  url_str STRING
)
RETURNS ARRAY<STRING>
DETERMINISTIC
LANGUAGE js
OPTIONS (
  library = ['gs://{bucket_name}/libs/uri-parser.js']
)
AS r"""
  if (!url_str) return [];
  try {
    return URIParser.getQueryKeys(url_str);
  } catch (err) {
    return [];
  }
""";

3. マネージド Python UDF（接続あり／なし）

Python UDF は永続化する必要があります（TEMP は使用しないでください）。 runtime_version = 'python-3.11' を指定し、entry_point を定義してください。 numpy、pandas、pyarrow は python-3.11 のベースイメージに プリインストールされています。

BigQuery のデフォルトは container_memory = '512Mi'、 container_cpu = 1.0、container_request_concurrency = 80 です。デフォルトで 十分な場合はこれらのオプションを省略し、ワークロードでコンテナのサイズや 同時実行数のカスタマイズが必要な場合にのみ明示的に設定してください（ Python UDFを参照）。

AS r'''...''' 内の Python のトップレベル文（import、ヘルパー関数、クラス、 および def {entry_point}(...)）は、先頭にインデントを付けず（左端に揃えて） 記述する必要があります。周囲の SQL に合わせてトップレベルの Python 定義を インデントすると、BigQuery がコンテナモジュールを読み込む際に IndentationError が発生します。

A. オフライン／ベクトル化 Python UDF（接続不要）

UDF が計算のみを実行する場合、または PyPI パッケージ（packages=[...]）／Cloud Storage ライブラリ（library=['gs://...']）を使用する場合は、WITH CONNECTION を省略します。 行のバッチに対する実行をベクトル化するには、パラメーターに pd.DataFrame（または pyarrow.RecordBatch）を型注釈として指定し、 pd.Series を返します。

警告

ベクトル化された操作は行ごとに独立している必要があります：BigQuery はベクトル化 UDF を任意の行バッチ（max_batching_rows まで）に対して呼び出します。df 全体を対象にした列の統計量（z スコア、データセットの平均／標準偏差、順位など）を計算しないでください。各バッチに含まれるのは、任意に選ばれた行の一部のみだからです。データセット全体の統計量には、SQL のウィンドウ関数（(val - AVG(val) OVER()) / NULLIF(STDDEV(val) OVER(), 0)）を使用してください。ベクトル化 Python UDF では、行ごとに独立した計算（幾何学的距離や複数列を使った計算など）を実行してください。

CREATE OR REPLACE FUNCTION `{project_id}.{dataset_id}.vectorized_distance`(
  x1 FLOAT64, y1 FLOAT64, x2 FLOAT64, y2 FLOAT64
)
RETURNS FLOAT64
LANGUAGE python
OPTIONS (
  runtime_version = 'python-3.11',
  entry_point = 'compute_distance'
)
AS r'''
import numpy as np
import pandas as pd

def compute_distance(df: pd.DataFrame) -> pd.Series:
    return pd.Series(
        np.hypot(df['x2'] - df['x1'], df['y2'] - df['y1']),
        index=df.index,
    )
''';

B. 外部／クラウド API を呼び出すオンライン Python UDF（WITH CONNECTION）

Python コードが実行時に外部ネットワークまたは API を呼び出す場合に限り、WITH CONNECTION を含めます。

try...except ブロックでは、必ずタイムアウト、ロギング、具体的な例外処理を設定してください。

• 一時的なエラーのみ → NULL：再試行可能な例外（google.api_core.exceptions.TooManyRequests、ServiceUnavailable、DeadlineExceeded、InternalServerError、RetryError）を捕捉し、logging.warning で記録してから None を返します。これにより、一時的な障害でクエリが中止されるのを防げます。ここでは基底クラスの GoogleAPICallError を捕捉しないでください。このクラスは PermissionDenied、Unauthenticated、InvalidArgument にも該当します。
• 致命的なエラー → 送出：認証、権限、無効な引数に関するエラーは伝播させます（またはログに記録して RuntimeError として再送出します）。これにより、エラーが黙って NULL を生成するのではなく、クエリのユーザーエラーとして表面化します。
• 行の値は決してログに記録しない：Cloud Logging にアクセスできるユーザーは、BigQuery のテーブル権限に関係なく UDF のログを読めます。入力データや出力データではなく、例外の種類とメッセージを記録してください。ログが出力されるのは、セッションで SET @@udf.properties = JSON '{"enableDebugOutput": true}'; が設定されている場合のみです。
• リクエストごとに timeout を設定する：Retry(timeout=...) が制限するのは再試行の合計時間のみです。各クライアント呼び出しにも timeout= を渡してください。

1. Cloud Resource Connection を作成し、そのサービスアカウントを取得する：

   bq mk --connection --connection_type=CLOUD_RESOURCE \
     --location={region} --project_id={project_id} {connection_id}
   
   bq show --connection {project_id}.{region}.{connection_id}
   

2. Connection のサービスアカウントに権限を付与する：{project_id} に roles/serviceusage.serviceUsageConsumer を付与し、さらに対象のプロジェクトに必要な API ロール（例：roles/cloudtranslate.user）を付与します。GRANT ... ON PROJECT の実行はプロジェクトの IAM ポリシーを変更するため、resourcemanager.projects.setIamPolicy（例：roles/resourcemanager.projectIamAdmin）が必要です。プロジェクトレベルの IAM 権限付与を実行する前に、ユーザーに確認してください。DCL の connection: の付与先にはサービスアカウントのメールアドレスではなく、Connection のパス（[project_id.]location.connection_id）を指定します。BigQuery が Connection のサービスアカウントを自動的に解決します（メールアドレスで付与する場合は "serviceAccount:{connection_sa_email}" を使用してください）。

   GRANT `roles/serviceusage.serviceUsageConsumer`
   ON PROJECT `{project_id}`
   TO "connection:{project_id}.{region}.{connection_id}";
   
   GRANT `roles/cloudtranslate.user`
   ON PROJECT `{project_id}`
   TO "connection:{project_id}.{region}.{connection_id}";
   

3. WITH CONNECTION を使用して Python UDF を作成する（Python のトップレベル文は列 0 から左詰めにしてください）：

CREATE OR REPLACE FUNCTION `{project_id}.{dataset_id}.translate_to_es`(
  text STRING
)
RETURNS STRING
LANGUAGE python
WITH CONNECTION `{project_id}.{region}.{connection_id}`
OPTIONS (
  entry_point = 'do_translate',
  runtime_version = 'python-3.11',
  packages = ['google-cloud-translate>=3.11', 'google-api-core']
)
AS r'''
import logging
from google.api_core import exceptions
from google.api_core.retry import Retry
from google.cloud import translate

client = translate.TranslationServiceClient()

TRANSIENT_ERRORS = (
    exceptions.TooManyRequests,
    exceptions.ServiceUnavailable,
    exceptions.DeadlineExceeded,
    exceptions.InternalServerError,
    exceptions.RetryError,
)

def do_translate(text: str) -> str:
    if not text:
        return None
    try:
        # Note: Replace {project_id} with your Google Cloud Project ID
        response = client.translate_text(
            request={
                "parent": "projects/{project_id}/locations/us-central1",
                "contents": [text],
                "target_language_code": "es",
                "mime_type": "text/plain",
            },
            retry=Retry(timeout=30.0),  # Total retry budget.
            timeout=10.0,  # Per-attempt deadline.
        )
        return response.translations[0].translated_text
    except TRANSIENT_ERRORS as e:
        # Log the error type only; never log row values.
        logging.warning("Transient Translation API error: %s", type(e).__name__)
        return None
    except exceptions.GoogleAPICallError as e:
        # PermissionDenied, Unauthenticated, InvalidArgument, etc. are fatal.
        logging.error("Fatal Translation API error: %s: %s", type(e).__name__, e)
        raise RuntimeError(f"Translation failed: {type(e).__name__}") from e
''';

コンテナのサイズ設定（container_cpu、container_memory、container_request_concurrency）、Cloud Logging（SET @@udf.properties = JSON '{"enableDebugOutput": true}';）、INFORMATION_SCHEMA.ROUTINES によるコンテナのビルド状態の確認については、Python UDF を参照してください。

4. リモート関数（Cloud Run エンドポイント）

エンドポイントが Cloud Run または Cloud Run functions でホストされている場合は、リモート関数を使用します（Python 以外の言語、VPC ネットワーキング、またはカスタムハードウェアの場合）。

1. 接続サービス アカウントに roles/run.invoker を付与します（Cloud Run サービスの IAM ポリシーを変更する前に、ユーザーに確認してください）。

   CLOUDSDK_METRICS_REQUEST_ATTRIBUTION="agent-skills/1.0 (skill:bigquery-udfs)" \
   gcloud run services add-iam-policy-binding {cloud_run_service} \
     --region={region} \
     --member="serviceAccount:{connection_sa_email}" \
     --role="roles/run.invoker"
   

2. リモート関数を登録します：エンドポイントは {"calls": [[arg1, ...], ...]} を受け取り、要素数が一致する {"replies": [res1, ...]} を返す必要があります。

   CREATE OR REPLACE FUNCTION `{project_id}.{dataset_id}.enrich_address`(
     street STRING, postal_code STRING
   )
   RETURNS JSON
   REMOTE WITH CONNECTION `{project_id}.{region}.{connection_id}`
   OPTIONS (
     endpoint = 'https://{cloud_run_service_url}',
     max_batching_rows = 1000
   );
   

Cloud Run ハンドラーのテンプレートと再試行の処理については、リモート関数を参照してください。

---

検証チェックリスト

UDF を生成または変更した後は、次の検証→修正→繰り返しの手順に従います。

• ステップ 1 — BigQuery DDL の前に Python コードを検証します：Python UDF の場合は、BigQuery DDL を送信する前に、Python 関数と packages の依存関係を検証します。ローカルで python3 のテスト入力（None、エッジケース、pd.DataFrame のバッチ）を使うか、コンテナ化された Cloud Run Job で検証します（gcloud run jobs deploy {job_name} --source . --region={region} と gcloud run jobs execute {job_name} --region={region} --wait を実行します。Python UDF の検証を参照してください）。

• ステップ 2 — DDL を実行し、コンテナのビルド状況をポーリングします：CREATE OR REPLACE FUNCTION を実行します。DDL ステートメントは Python コンテナイメージのビルド完了前に返されるため、build_state が IN_PROGRESS から SUCCEEDED に変わるまで INFORMATION_SCHEMA.ROUTINES をポーリングします（FAILED の場合は修正して再実行します）。

  SELECT build_status.*
  FROM `{project_id}.{dataset_id}`.INFORMATION_SCHEMA.ROUTINES
  WHERE routine_name = '{udf_name}';
  

• ステップ 3 — SQL の SELECT で UDF をテストします：build_state = 'SUCCEEDED' になったら（SQL/JS UDF の場合は DDL の直後に）、NULL と有効な入力を網羅するテスト用の SELECT クエリを実行します。

クリーンアップと削除

使用していないルーティン、接続、Cloud Run サービス、ステージング済みの Cloud Storage アーティファクトを削除します（プロジェクトおよびリージョンごとに、すべての Python UDF コンテナの合計サイズは 10 GiB に制限されていることに注意してください）。

bq query --use_legacy_sql=false "DROP FUNCTION IF EXISTS \`{project_id}.{dataset_id}.{udf_name}\`;"
bq rm --connection --location={region} {connection_id}
CLOUDSDK_METRICS_REQUEST_ATTRIBUTION="agent-skills/1.0 (skill:bigquery-udfs)" \
gcloud run services delete {cloud_run_service} --region={region} --project={project_id}
CLOUDSDK_METRICS_REQUEST_ATTRIBUTION="agent-skills/1.0 (skill:bigquery-udfs)" \
gcloud storage rm gs://{bucket_name}/udf_libs/bundle.js

リファレンスディレクトリ

• SQL と JavaScript の UDF：テンプレート化された SQL パラメーター（ANY TYPE）、集約 UDF、JavaScript の型マッピング、Cloud Storage ライブラリの読み込み (https://docs.cloud.google.com/bigquery/docs/user-defined-functions.md.txt)。
• Python UDF：オフラインと WITH CONNECTION によるオンラインサービスアクセス、Pandas/PyArrow のベクトル化、コンテナのリソース 制限（container_cpu/container_memory）、Cloud Logging によるデバッグ (https://docs.cloud.google.com/bigquery/docs/user-defined-functions-python.md.txt)。
• リモート関数：Cloud Run の HTTP リクエストと レスポンスの JSON スキーマ、バッチサイズ、再試行のセマンティクス、VPC のエグレス (https://docs.cloud.google.com/bigquery/docs/remote-functions.md.txt)。
• IAM、承認済みルーティン、割り当て： データセット間でのルーティンの共有（Authorized Routines）、コンテナの割り当て、 エラーのトラブルシューティング。

関連スキル

• BigQuery Basics（bigquery-basics）：データセット、テーブル、ジョブの管理。
• BigQuery DataFrames：@bpd.udf を使用した Python による Python UDF の作成。
• BigQuery AI & ML：組み込みの ML.GENERATE_TEXT および AI.GENERATE 関数。