解决BigQuery将空数组Parquet的list<string>识别为list<int32>问题

我有若干TB级的嵌套JSONL文件，正在将其转换为Parquet文件并写入分区的Google Cloud Storage（GCS）存储桶。遇到以下问题：

• 嵌套字段billing_code_modifier预期架构为list<item: string>
• 当所有记录中该列表长度均为0时，Pandas会将其写入为list<item: null>
• BigQuery读取这类Parquet文件时会失败，因为BigQuery将空数组默认识别为int32类型，引发架构不一致
• 受限于GCS特性，无法先写入空Parquet再追加数据，仅允许覆盖文件

1. 显式指定Parquet写入架构（推荐） #

通过pyarrow定义明确的字段类型，强制billing_code_modifier为字符串列表类型，即使数组为空也能保留正确的架构，从根源避免类型识别错误。

实现代码 #

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

# 加载JSONL数据（支持直接读取GCS路径）
df = pd.read_json("gs://your-bucket/input/*.jsonl", lines=True)

# 定义完整的目标架构，替换为你的实际字段类型
target_schema = pa.schema([
    # pa.field("order_id", pa.int64()),
    # pa.field("total_amount", pa.float64()),
    pa.field("billing_code_modifier", pa.list_(pa.string()))
])

# 将DataFrame转换为pyarrow Table并应用指定架构
table = pa.Table.from_pandas(df, schema=target_schema)

# 写入分区化Parquet到GCS
pq.write_to_dataset(
    table,
    root_path="gs://your-bucket/output/partitioned_data/",
    partition_cols=["your_partition_column"],  # 替换为你的分区字段
    schema=target_schema,
    overwrite=True
)

2. 预处理空列表（临时兼容方案） #

如果不想引入pyarrow，可以在Pandas层面预处理数据：当检测到billing_code_modifier全为空列表时，给单条记录的该字段注入一个空字符串，强制Pandas识别为字符串列表类型。

实现代码 #

import pandas as pd

df = pd.read_json("gs://your-bucket/input/*.jsonl", lines=True)

# 检查所有记录的billing_code_modifier是否都是空列表
all_empty = df["billing_code_modifier"].apply(lambda x: len(x) == 0).all()

if all_empty:
    # 给第一条记录注入空字符串，强制类型为string列表
    df.at[0, "billing_code_modifier"] = [""]

# 写入Parquet到GCS
df.to_parquet(
    "gs://your-bucket/output/partitioned_data/",
    engine="pyarrow",
    partition_cols=["your_partition_column"],
    overwrite=True
)

注意：该方法会修改原始数据（注入一个空字符串），需确认业务上可接受后使用。

3. 用Dask处理TB级大文件 #

针对TB级数据，单进程Pandas可能存在内存瓶颈，使用Dask可并行处理数据，同时支持指定pyarrow架构保证类型正确。

实现代码 #

import dask.dataframe as dd
import pyarrow as pa

# 并行读取GCS上的JSONL文件
ddf = dd.read_json("gs://your-bucket/input/*.jsonl", lines=True)

# 定义目标架构
target_schema = pa.schema([
    pa.field("billing_code_modifier", pa.list_(pa.string()))
])

# 写入分区化Parquet到GCS
ddf.to_parquet(
    "gs://your-bucket/output/partitioned_data/",
    engine="pyarrow",
    schema=target_schema,
    partition_on=["your_partition_column"],
    overwrite=True
)

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