bigquery-udfs

BigQuery 用户定义函数（UDF）

BigQuery 用户定义函数（UDF）通过以 SQL、JavaScript、托管 Python（python-3.11）编写的自定义逻辑，或通过托管在 Cloud Run 和 Cloud Run functions 上的远程函数，扩展 GoogleSQL。

归因

使用此技能时，直接在终端中执行的所有 BigQuery 或 Google Cloud 命令都必须添加标记，以便进行使用归因。该标记仅标识技能及其版本，不包含任何用户数据。

• 无论子命令是什么，每次调用 gcloud 时都要在命令前添加指标环境变量。每条命令都要内联设置这些变量；Shell 状态可能不会在命令之间保留：

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

  不要为此使用 gcloud config set：它会在当前任务结束后继续生效，并错误标记无关的使用情况。

• 直接调用 BigQuery API（例如使用 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 wheel（packages=[...]）、Cloud Storage .py 模块（library=[...]）或向量化 Pandas／PyArrow 批次执行纯计算。	仅支持持久化函数（不支持 CREATE TEMP FUNCTION）。禁止出站网络访问。容器规格上限：4 vCPU／16 GiB。不支持 JSON、RANGE、INTERVAL、GEOGRAPHY。
托管式 Python UDF（WITH CONNECTION）	是（CLOUD_RESOURCE）	从 BigQuery 容器内的 Python 代码调用 Google Cloud API（例如 Cloud Translation、Gemini Enterprise Agent Platform）或在线 HTTP 端点。	必须使用 WITH CONNECTION。在 VPC-SC 周界内，出站流量仅允许访问 restricted.googleapis.com（不支持 VPC 网络）。容器规格上限：4 vCPU／16 GiB。
远程函数（REMOTE WITH CONNECTION）	是（CLOUD_RESOURCE）	调用外部 API 或微服务，且需要非 Python 运行时（Go、Java、Node.js）、私有 VPC 网络访问、GPU 或自定义 Cloud Run 扩缩容时。	需要实现 BigQuery JSON 批次协议的 Cloud Run HTTPS 端点，并配置 roles/run.invoker。每行输入最大 5 MB；响应大小上限为 15 MB（Cloud Run）／10 MB（第 1 代）。

前提条件与设置

确保 Google Cloud 项目已关联有效的结算账号（参见 BigQuery 定价）：

1. 安装并验证 Google Cloud SDK：安装 Google Cloud SDK 并进行身份验证：

   gcloud auth login
   

   在每条命令中显式传入 --project={project_id} 和 --region={region}（对于 bq，传入 --location），不要运行 gcloud config set，因为其设置会在任务结束后继续保留。如果用户未指定区域，默认使用 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（通过 GRANT ... ON PROJECT 向连接服务账号授予项目级 IAM 角色时，还需要 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}(...)）必须完全不缩进 （从最左侧边界开始）。将顶层 Python 定义缩进以与周围的 SQL 对齐，会导致 BigQuery 加载容器模块时出现 IndentationError。

A. 离线 / 向量化 Python UDF（无需连接）

如果 UDF 仅执行计算，或使用 PyPI 包（packages=[...]）/ Cloud Storage 库（library=['gs://...']），请省略 WITH CONNECTION。 若要对行批次执行向量化处理，请将参数标注为 pd.DataFrame（或 pyarrow.RecordBatch），并返回 pd.Series：

警告

向量化操作必须与行相互独立：BigQuery 会对任意行批次（最多 max_batching_rows 行）调用 向量化 UDF。请勿跨 df 计算全局列统计量（例如 z-score、数据集均值/标准差或排名）， 因为每个批次只能看到任意行切片。对于整个数据集范围的统计量，请使用 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. 在线 Python UDF 调用外部 / 云 API（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 访问权限的人都可以读取 UDF 日志， 无论其是否拥有 BigQuery 表权限。记录异常类型和消息，不要记录输入或输出 数据。只有会话设置了 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
''';

请参阅 Python UDFs，了解容器大小设置 （container_cpu、container_memory、container_request_concurrency）、Cloud Logging（SET @@udf.properties = JSON '{"enableDebugOutput": true}';），以及 如何通过 INFORMATION_SCHEMA.ROUTINES 检查容器构建状态。

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 容器镜像构建完成之前返回，请轮询 INFORMATION_SCHEMA.ROUTINES，直到 build_state 从 IN_PROGRESS 转为 SUCCEEDED（如果状态为 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 执行后立即测试），运行测试 SELECT 查询，覆盖 NULL 和有效输入。

清理与拆除

删除未使用的例程、连接、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 基础知识（bigquery-basics）：数据集、表和作业管理。
• BigQuery DataFrames：通过 Python 中的 @bpd.udf 创建 Python UDF。
• BigQuery AI & ML：内置的 ML.GENERATE_TEXT 和 AI.GENERATE 函数。