在Python函数中执行参数化BigQuery SQL时遇语法错误求助

错误信息 #

BadRequest: 400 1.2 - 1.118: Unrecognized token CREATE.
[Try using standard SQL]

(job ID: 7417d5d6-fdcd-420e-b7ac-4aaa8bb3347c)

-----Query Job SQL Follows-----                                               

|    .    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |

1: CREATE OR REPLACE TABLE analytics-mkt-cleanroom.MKT_DS.PXV2DWY_HS_MODEL_INTRMDT_TAB_01 AS SELECT '2021-07-01' AS DT
| . | . | . | . | . | . | . | . | . | . | . |

完整代码 #

# Creating and initializing a random table:
%%bigquery
CREATE OR REPLACE TABLE `analytics-mkt-cleanroom.MKT_DS.Home_Services_PXV2DWY_HS_MODEL_INTRMDT_TABLE_01` AS 
SELECT CURRENT_DATE AS DT

# Checking what's the current date:
%%bigquery
SELECT * FROM `analytics-mkt-cleanroom.MKT_DS.Home_Services_PXV2DWY_HS_MODEL_INTRMDT_TABLE_01`

# Initializing random str date variable:
from_date = f"'2021-07-01'"
to_date = f"'2022-06-30'"

# Creating a Python function to update the existing table using a parameter:
from google.cloud import bigquery
def my_func(from_date):
    client = bigquery.Client(project='analytics-mkt-cleanroom')
    job_config = bigquery.QueryJobConfig()
    job_config.use_legacy_sql = True   

    destination_table_id = f'`analytics-mkt-cleanroom.MKT_DS.PXV2DWY_HS_MODEL_INTRMDT_TAB_01`'
    
    sql = """ CREATE OR REPLACE TABLE """ + destination_table_id + """ AS SELECT {0} AS DT """.format(from_date)
    query = client.query(sql, job_config=job_config)
    query.result()
    return

# Checking what's the SQL that is getting generated inside:
destination_table_id = f'`analytics-mkt-cleanroom.MKT_DS.PXV2DWY_HS_MODEL_INTRMDT_TAB_01`'
sql = """ CREATE OR REPLACE TABLE """ + destination_table_id + """ AS SELECT {0} AS DT """.format(from_date)
sql

my_func(from_date)

问题原因 #

Python函数里显式设置了job_config.use_legacy_sql = True，但CREATE OR REPLACE TABLE是BigQuery标准SQL的语法，Legacy SQL不支持该语法，因此触发"Unrecognized token CREATE"错误。

修复方案 #

1. 切换为标准SQL #

将job_config.use_legacy_sql = True修改为job_config.use_legacy_sql = False，或者直接删除这一行（BigQuery客户端默认使用标准SQL）：

def my_func(from_date):
    client = bigquery.Client(project='analytics-mkt-cleanroom')
    job_config = bigquery.QueryJobConfig()
    # 修改或移除该行
    job_config.use_legacy_sql = False   

    destination_table_id = f'`analytics-mkt-cleanroom.MKT_DS.PXV2DWY_HS_MODEL_INTRMDT_TAB_01`'
    
    sql = """ CREATE OR REPLACE TABLE """ + destination_table_id + """ AS SELECT {0} AS DT """.format(from_date)
    query = client.query(sql, job_config=job_config)
    query.result()
    return

2. 推荐：使用参数化查询避免SQL注入 #

直接字符串拼接参数存在SQL注入风险，建议使用BigQuery参数化查询功能：

def my_func(from_date):
    client = bigquery.Client(project='analytics-mkt-cleanroom')
    job_config = bigquery.QueryJobConfig()
    
    # 定义查询参数
    job_config.query_parameters = [
        bigquery.ScalarQueryParameter("from_date", "STRING", from_date.strip("'"))
    ]
    
    destination_table_id = '`analytics-mkt-cleanroom.MKT_DS.PXV2DWY_HS_MODEL_INTRMDT_TAB_01`'
    # 使用@占位符引用参数
    sql = f""" CREATE OR REPLACE TABLE {destination_table_id} AS SELECT @from_date AS DT """
    
    query = client.query(sql, job_config=job_config)
    query.result()
    return

同时需要调整日期变量的初始化（去掉单引号）：

from_date = "2021-07-01"
to_date = "2022-06-30"

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