通过Airflow处理BigQuery查询结果中的未知字符�问题

报错根因为BigQuery Storage v1 API 默认开启严格UTF-8解码，遇到不符合规范的字节序列时会触发序列化异常，BQ控制台使用宽松解码策略因此可正常展示结果。以下三种方案均可在你现有代码的第三行及之后调整生效：

方案1：临时禁用BigQuery Storage API（小数据集首选，改动最小） #

直接在to_dataframe方法中关闭Storage API调用，降级为传统REST接口拉取数据，默认使用宽松编码处理：

df = bq_client.query(SQL_QUERY).to_dataframe(use_bqstorage_api=False)

注意：该方案会拉低大数据量的查询转换性能，单表数据量大于10G时不推荐使用。

方案2：配置Storage API解码策略（大数据集首选，保留高性能） #

自定义BigQuery Storage客户端的序列化参数，指定遇到非法字节时自动忽略或替换，不影响Storage API的传输性能：

from google.cloud import bigquery_storage_v1
from google.cloud.bigquery_storage_v1 import types

# 初始化带自定义配置的Storage客户端
bq_storage_client = bigquery_storage_v1.BigQueryReadClient(credentials=bq_hook._get_credentials())
# 配置解码错误处理规则：IGNORE为忽略非法字节，REPLACE为替换为�字符
read_options = types.ReadSession.TableReadOptions(
    arrow_serialization_options=types.ArrowSerializationOptions(
        buffer_error_handling=types.ArrowSerializationOptions.BufferErrorHandling.IGNORE
    )
)

# 传入自定义配置完成数据转换
df = bq_client.query(SQL_QUERY).to_dataframe(
    bqstorage_client=bq_storage_client,
    bqstorage_read_options=read_options
)

方案3：手动处理Arrow结果编码（极端场景使用，灵活性最高） #

如果上述两种方案仍无法匹配你的业务需求，可以先将查询结果转为Arrow中间表，手动处理所有字符串列的编码后再转为Pandas DataFrame：

import pyarrow

# 先获取Arrow格式查询结果
arrow_table = bq_client.query(SQL_QUERY).to_arrow()

# 遍历所有字符串列处理编码问题
for col_name in arrow_table.column_names:
    if arrow_table.schema.field(col_name).type == pyarrow.string():
        processed_col = pyarrow.array(
            [s.decode("utf-8", errors="replace") if isinstance(s, bytes) else s 
             for s in arrow_table[col_name].to_pylist()],
            type=pyarrow.string()
        )
        arrow_table = arrow_table.set_column(
            arrow_table.schema.get_field_index(col_name),
            col_name,
            processed_col
        )

# 转换为最终DataFrame
df = arrow_table.to_pandas()

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