从BigQuery存储过程调用第二代Cloud Function报错及最佳方式咨询

错误原因分析 #

你遇到的Unexpected TVF argument type found, expecting literal, query parameter, or variable of type STRING错误，根源在于两个误用：

1. CONNECTION_ID函数的错误用法：该函数仅用于引用BigQuery已创建的外部连接资源（比如连接Cloud SQL、AWS Athena的预配置连接），不能直接传入Cloud Function的URL。
2. EXTERNAL_QUERY的适用场景错误：这个函数是用来查询外部数据源的，并非用于调用HTTP服务（比如Cloud Function）。

BigQuery调用Cloud Function的正确方式 #

BigQuery无法直接通过存储过程发起HTTP请求调用Cloud Function，需要通过以下两种方案实现：

方案1：使用外部UDF（同步获取结果） #

这是需要在存储过程中同步获取Cloud Function返回结果的最佳方式，步骤如下：

1. 配置Cloud Function

确保你的第二代Cloud Function是HTTP触发器，并且返回符合BigQuery外部UDF要求的JSON格式。示例Python代码：

import json

def process(request):
    # 处理BigQuery传入的请求（如果需要参数）
    request_data = request.get_json()
    
    # 执行你的业务逻辑
    result = "Cloud Function执行完成"
    
    # 返回BigQuery要求的格式：包含schema和rows
    return json.dumps({
        "schema": [{"name": "result", "type": "STRING"}],
        "rows": [[result]]
    })

2. 创建BigQuery外部连接

在BigQuery中创建一个Cloud Resource连接，区域选择和Cloud Function一致的europe-west1，并确保该连接的服务账号拥有roles/cloudfunctions.invoker权限（能调用目标Cloud Function）。

3. 创建外部UDF

在BigQuery中创建指向Cloud Function的外部UDF：

CREATE OR REPLACE FUNCTION `project-dev.dataset.call_process_cf`()
RETURNS STRING
REMOTE WITH CONNECTION `project-dev.europe-west1.your_connection_name` -- 替换为你的连接名
OPTIONS (
    endpoint = 'https://europe-west1-project-dev.cloudfunctions.net/process',
    user_defined_context = {"env": "dev"} -- 可选：传递自定义上下文参数
);

4. 在存储过程中调用UDF

修改你的存储过程，直接调用创建好的外部UDF：

CREATE OR REPLACE PROCEDURE `project-dev.dataset.testCF`()
OPTIONS(strict_mode=false)
BEGIN
  DECLARE response STRING;
  SET response = `project-dev.dataset.call_process_cf`();
  SELECT response;
END;

方案2：使用Pub/Sub触发（异步执行） #

如果不需要在存储过程中同步获取结果，可以通过Pub/Sub间接触发Cloud Function：

• 创建一个Pub/Sub主题，让你的Cloud Function订阅该主题
• 在BigQuery中通过数据导出到Pub/Sub，或者编写一个辅助Cloud Function，在存储过程中通过写入特定表触发辅助函数发送Pub/Sub消息，进而触发目标Cloud Function

总结 #

• 不能直接用EXTERNAL_QUERY调用Cloud Function，该函数仅适用于外部数据源查询
• 同步场景优先选择外部UDF方案，异步场景可选择Pub/Sub触发方案

内容的提问来源于stack exchange，提问作者mikestr