Python迁移RedShift含JSON列表至BigQuery报错，求正确Schema配置

问题1：使用JSON类型时的"cannot mix list and non-list, non-null values"错误 #

原因 #

BigQuery的JSON类型要求字段值统一为JSON格式的字符串，你的DataFrame中log字段可能混合了Python字典/列表对象、空值（None）或非JSON字符串，导致类型不兼容。

解决步骤 #

1. 统一转换log列为JSON字符串格式，同时处理空值：

import json

# 在调用load_data函数前，预处理DataFrame的log列
data['log'] = data['log'].apply(
    lambda x: json.dumps(x) if x is not None else "null"
)

2. 保持原Schema中log字段的类型为JSON不变，重新执行加载即可。

问题2：使用RECORD类型时的"was expecting tuple of (key, value) pair"错误 #

原因 #

你的Schema存在多处语法错误：

• 嵌套SchemaField定义后缺少逗号分隔
• fields参数的位置错误（应作为SchemaField的参数传入括号内，而非单独写在外面）
• 字段名的引号拼写错误（比如"address,"多了逗号）
• 部分字段类型匹配错误（比如number是数字，不应设为STRING）

正确的嵌套RECORD Schema写法 #

def load_data(table_id, data):
    print("load_data::Writing records to table", table_id)
    # 预处理log列：如果是JSON字符串，先解析为Python字典
    import json
    data['log'] = data['log'].apply(lambda x: json.loads(x) if x is not None else None)
    
    job_config = bigquery.LoadJobConfig(
        write_disposition="WRITE_APPEND",
        schema=[
            bigquery.SchemaField("id", "INT64"),
            bigquery.SchemaField("user_id", "INT64"),
            bigquery.SchemaField(
                "log", "RECORD", fields=[
                    bigquery.SchemaField(
                        "reason", "RECORD", fields=[
                            bigquery.SchemaField("id", "INT64"),
                            bigquery.SchemaField("name", "STRING"),
                            bigquery.SchemaField(
                                "contact", "RECORD", fields=[
                                    bigquery.SchemaField("number", "INT64"),
                                    bigquery.SchemaField("address", "STRING")
                                ]
                            )
                        ]
                    ),
                    bigquery.SchemaField(
                        "subreason", "RECORD", fields=[
                            bigquery.SchemaField("id", "INT64"),
                            bigquery.SchemaField("name", "STRING"),
                            bigquery.SchemaField("is_active", "BOOLEAN"),
                            bigquery.SchemaField("created_at", "TIMESTAMP"),
                            bigquery.SchemaField("deleted_at", "TIMESTAMP"),
                            bigquery.SchemaField("complaint_id", "INT64")
                        ]
                    )
                ]
            )
        ]
    )
    try:
        start = time.time()
        job = client.load_table_from_dataframe(
            data, table_id, job_config=job_config
        )
        job.result()
        end = time.time()
        print("load_data::Time taken for writing " + str(data.shape[0]) + " records: ", end - start, "s")
    except Exception as e:
        print("load_data::exception", e)
        print("load_data::Could not establish connection with Google BigQuery. Terminating program")
        conn.close()
        sys.exit()

两种方案对比 #

• JSON类型：无需定义复杂嵌套结构，适合JSON格式不固定或经常变化的场景，但查询时需要用BigQuery的JSON函数（如JSON_EXTRACT）解析字段。
• RECORD类型：结构化存储，支持直接用SQL访问嵌套字段，查询效率更高，但要求JSON结构固定，后续结构变化时需要同步修改Schema。

内容的提问来源于stack exchange，提问作者Khawaja Abdul Ahad