将字典列表插入BigQuery临时表遇语法错误，求可行解决方案

问题场景 #

拥有以表列为键的字典列表，示例如下：

[
    {'mmsi': '256718000', 'callsign': '9HA3989', 'imo_lr_ihs_no': '1006532', 'name_of_ship': 'SEAHORSE', 'ship_type': 'Yacht', 'gt': '610', 'group_owner': 'Taracan Investments SA', 'technical_manager': 'Alpha + Ltd', 'shipmanager': 'Alpha + Ltd', 'registered_owner': 'Seahorse Navigation Ltd', 'operator': 'Alpha + Ltd', 'operator_code': '5733441', 'registered_owner_code': '6396829', 'length': '51.690'},
    {'mmsi': '319833000', 'callsign': '9HA5138', 'imo_lr_ihs_no': '1006673', 'name_of_ship': 'SENSES', 'ship_type': 'Yacht', 'gt': '993', 'group_owner': '  Unknown', 'technical_manager': 'YCO SAM', 'shipmanager': 'YCO SAM', 'registered_owner': 'Argo Marine Ltd', 'operator': 'YCO SAM', 'operator_code': '5073856', 'registered_owner_code': '6126096', 'length': '57.000'}
]

尝试用以下SQL方式创建临时表：

query = f'CREATE TEMPORARY TABLE temp_table AS SELECT * FROM UNNEST({all_ship_info})'
print(query)
result = client.query(query)
print('Result', result.result())

执行后出现错误：

google.api_core.exceptions.BadRequest: 400 Braced constructors are not supported at [1:60]

错误原因 #

直接将Python格式的字典列表字符串插入BigQuery SQL中，BigQuery不支持Python这种{'key':'value'}的字典构造语法，需要转换为BigQuery原生支持的STRUCT数组格式。

解决方案 #

方法一：转换字典列表为BigQuery STRUCT数组语法 #

手动将每个Python字典转换为BigQuery的STRUCT(字段名='值', ...)格式，再组合成数组字符串，示例代码：

def convert_to_bq_struct_array(dict_list):
    structs = []
    for item in dict_list:
        # 处理值中的单引号，避免SQL语法错误
        escaped_fields = [f"{k}='{v.replace(\"'\", \"''\")}'" for k, v in item.items()]
        struct_str = f"STRUCT({', '.join(escaped_fields)})"
        structs.append(struct_str)
    return f"[{', '.join(structs)}]"

# 转换你的字典列表
bq_array_str = convert_to_bq_struct_array(all_ship_info)

# 构建并执行查询
query = f'CREATE TEMPORARY TABLE temp_table AS SELECT * FROM UNNEST({bq_array_str})'
result = client.query(query)
result.result()

方法二：使用参数化查询（推荐，更安全） #

通过BigQuery的参数化查询传递STRUCT数组，避免字符串拼接的SQL注入风险，也无需手动转义特殊字符：

from google.cloud import bigquery
from google.cloud.bigquery import ArrayQueryParameter, StructQueryParameter

client = bigquery.Client()

# 定义STRUCT的字段和对应数据类型，根据实际字段类型调整
struct_field_defs = [
    StructQueryParameter("mmsi", "STRING"),
    StructQueryParameter("callsign", "STRING"),
    StructQueryParameter("imo_lr_ihs_no", "STRING"),
    StructQueryParameter("name_of_ship", "STRING"),
    StructQueryParameter("ship_type", "STRING"),
    StructQueryParameter("gt", "STRING"),
    StructQueryParameter("group_owner", "STRING"),
    StructQueryParameter("technical_manager", "STRING"),
    StructQueryParameter("shipmanager", "STRING"),
    StructQueryParameter("registered_owner", "STRING"),
    StructQueryParameter("operator", "STRING"),
    StructQueryParameter("operator_code", "STRING"),
    StructQueryParameter("registered_owner_code", "STRING"),
    StructQueryParameter("length", "STRING"),
]

# 构造数组参数
struct_type = f"STRUCT<{', '.join([f'{p.name}:{p.parameter_type}' for p in struct_field_defs])}>"
query_params = [
    ArrayQueryParameter("ship_info_array", struct_type, all_ship_info)
]

# 配置查询作业
job_config = bigquery.QueryJobConfig(query_parameters=query_params)
query = """
CREATE TEMPORARY TABLE temp_table AS
SELECT * FROM UNNEST(@ship_info_array)
"""

# 执行查询
result = client.query(query, job_config=job_config)
result.result()

后续合并现有表 #

创建临时表后，即可用MERGE语句合并到目标表，示例：

MERGE INTO `your-project.your-dataset.target_table` AS target
USING temp_table AS source
ON target.mmsi = source.mmsi
WHEN MATCHED THEN
  UPDATE SET
    callsign = source.callsign,
    name_of_ship = source.name_of_ship,
    -- 其他需要更新的字段
WHEN NOT MATCHED THEN
  INSERT (mmsi, callsign, name_of_ship, ...)
  VALUES (source.mmsi, source.callsign, source.name_of_ship, ...)

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