使用load_table_from_dataframe导入含JSON列的DataFrame到BigQuery失败求助

尝试将包含JSON列的Pandas DataFrame加载到BigQuery时，load_table_from_dataframe方法不支持BigQuery原生JSON类型，即使手动设置Schema或禁用自动检测也无法解决，报错Unsupported field type: JSON。

目标表SQL #

create table `project_id`.sandbox.tmp (id int, json_data json);

测试代码 #

import pandas as pd
from google.cloud import bigquery

# 创建包含JSON列的Pandas DataFrame
df = pd.DataFrame({
        'id': [1, 2, 3],
        'json_data': [{"key": "value1"}, {"key": "value2"}, {"key": "value3"}]
    })

client = bigquery.Client()
table_ref = "project_id.sandbox.tmp"
job_config = bigquery.LoadJobConfig()
job_config.autodetect = False # 无效果
# job_config.autodetect = True

job = client.load_table_from_dataframe(df, table_ref, job_config=job_config)
job.result()

报错信息 #

/...-service/venv/lib/python3.11/site-packages/google/cloud/bigquery/_pandas_helpers.py:267: UserWarning: Unable to determine type for field 'json_data'.
  warnings.warn("Unable to determine type for field '{}'.".format(bq_field.name))
Traceback (most recent call last):
  File "....service/other/sample_load_from_dataframe.py", line 21, in <module>
    job.result()
  File "/....-service/venv/lib/python3.11/site-packages/google/cloud/bigquery/job/base.py", line 922, in result
    return super(_AsyncJob, self).result(timeout=timeout, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/....-service/venv/lib/python3.11/site-packages/google/api_core/future/polling.py", line 261, in result
    raise self._exception
google.api_core.exceptions.BadRequest: 400 Unsupported field type: JSON

环境信息 #

• Python: 3.11
• google-cloud-bigquery: 3.11.4

方法1：将JSON列序列化为字符串后加载 #

把DataFrame中的JSON对象转为JSON字符串，加载时指定列类型为STRING，BigQuery会自动将合法的JSON字符串转换为原生JSON类型（目标表列需为JSON类型）。

import pandas as pd
import json
from google.cloud import bigquery

df = pd.DataFrame({
    'id': [1, 2, 3],
    # 将JSON对象序列化为字符串
    'json_data': [json.dumps(item) for item in [{"key": "value1"}, {"key": "value2"}, {"key": "value3"}]]
})

client = bigquery.Client()
table_ref = "project_id.sandbox.tmp"

job_config = bigquery.LoadJobConfig(
    schema=[
        bigquery.SchemaField("id", "INTEGER"),
        # 显式指定为STRING类型
        bigquery.SchemaField("json_data", "STRING")
    ]
)

job = client.load_table_from_dataframe(df, table_ref, job_config=job_config)
job.result()

方法2：通过Parquet中间文件加载 #

利用Parquet格式支持结构化数据的特性，先将DataFrame导出为Parquet字节流，再用load_table_from_file加载到BigQuery。

import pandas as pd
from google.cloud import bigquery
import io

df = pd.DataFrame({
    'id': [1, 2, 3],
    'json_data': [{"key": "value1"}, {"key": "value2"}, {"key": "value3"}]
})

# 将DataFrame转为Parquet字节流
parquet_buffer = io.BytesIO()
df.to_parquet(parquet_buffer, engine='pyarrow')
parquet_buffer.seek(0)

client = bigquery.Client()
table_ref = "project_id.sandbox.tmp"

job_config = bigquery.LoadJobConfig(
    source_format=bigquery.SourceFormat.PARQUET,
    schema=[
        bigquery.SchemaField("id", "INTEGER"),
        bigquery.SchemaField("json_data", "JSON")
    ]
)

job = client.load_table_from_file(parquet_buffer, table_ref, job_config=job_config)
job.result()

方法3：直接执行BigQuery INSERT语句 #

遍历DataFrame构建INSERT语句，将JSON对象转为字符串后插入目标表。

import pandas as pd
import json
from google.cloud import bigquery

df = pd.DataFrame({
    'id': [1, 2, 3],
    'json_data': [{"key": "value1"}, {"key": "value2"}, {"key": "value3"}]
})

client = bigquery.Client()
table_ref = "project_id.sandbox.tmp"

# 构建批量INSERT的VALUES部分
values_list = []
for _, row in df.iterrows():
    json_str = json.dumps(row['json_data']).replace("'", "\\'")  # 转义单引号避免SQL语法错误
    values_list.append(f"({row['id']}, '{json_str}')")

# 拼接完整INSERT语句
insert_query = f"""
INSERT INTO `{table_ref}` (id, json_data)
VALUES {', '.join(values_list)}
"""

query_job = client.query(insert_query)
query_job.result()

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