Airflow BigQueryOperator调用UDF失败问题求助

我来帮你搞定这个「Function not found: multiplyInputs」的报错！你的核心问题出在UDF定义的语法错误和udf_config的格式细节上，咱们一步步修正：

问题根源 #

你当前的udf_config存在两个关键问题：

1. JS代码块的引号嵌套错误：你尝试用\转义，但结合三重引号的写法导致UDF定义不完整，BigQuery无法正确解析并注册这个函数。
2. 字符串末尾多余的无效语法：""; ""这部分属于冗余内容，直接破坏了整个UDF的DDL语句结构。

修正方案 #

1. 简化UDF的语法格式 #

BigQuery的JS临时UDF可以直接用单引号包裹JS代码块，这样能避免Python字符串中嵌套三重引号的冲突。正确的UDF定义如下：

CREATE TEMPORARY FUNCTION multiplyInputs(x FLOAT64, y FLOAT64)
RETURNS FLOAT64
LANGUAGE js
AS 'return x*y;';

2. 正确配置udf_config #

udf_config确实要求是列表类型，每个元素对应一个完整的UDF DDL语句。把上面修正后的UDF定义作为列表的唯一元素即可。

完整修正后的代码 #

import datetime
from airflow import models
from airflow.contrib.operators import bigquery_operator

yesterday = datetime.datetime.combine(datetime.datetime.today() - datetime.timedelta(1), datetime.datetime.min.time())

default_dag_args = {
    'start_date': yesterday,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': 'vital-platform-791'
}

with models.DAG('udf_example', schedule_interval=datetime.timedelta(days=1), default_args=default_dag_args) as dag:
    table = 'udf_table'
    task_id = table + '_fe'
    
    # 修正后的udf_config：语法正确的UDF定义作为列表元素
    udf_config = ["""CREATE TEMPORARY FUNCTION multiplyInputs(x FLOAT64, y FLOAT64)
                  RETURNS FLOAT64
                  LANGUAGE js
                  AS 'return x*y;';"""]
    
    query = """WITH numbers AS (
                  SELECT 1 AS x, 5 as y UNION ALL
                  SELECT 2 AS x, 10 as y UNION ALL
                  SELECT 3 as x, 15 as y)
                SELECT x, y, multiplyInputs(x, y) as product FROM numbers"""
    
    destination_table = 'vital-platform-791.alpha_factors.{table}_fe'.format(table=table)
    t_fe = bigquery_operator.BigQueryOperator(
        task_id=task_id,
        bql=query,
        destination_dataset_table=destination_table,
        use_legacy_sql=False,
        write_disposition='WRITE_TRUNCATE',
        udf_config=udf_config
    )

额外注意事项 #

• 务必保持use_legacy_sql=False（你已经正确设置），因为BigQuery的JS UDF仅支持标准SQL语法。
• 如果后续升级Airflow版本，建议改用apache-airflow-providers-google包中的BigQueryOperator，它的API更稳定，udf_config的用法与当前保持一致。
• 建议先在BigQuery控制台中测试UDF+查询的组合，确认能正常执行后再复制到Airflow代码中，提前规避语法问题。

内容的提问来源于stack exchange，提问作者Richardt Benade REZCO