使用Apache Beam写入BigQuery时遇AttributeError问题排查

这个错误的核心原因是：代码中某个逻辑试图对整数（int）对象调用字符串的encode方法，而这是不允许的。结合报错发生在WriteToBigQuery/Map阶段，可按以下步骤排查：

1. 核对BigQuery表Schema与JSON数据的类型匹配 #

• 检查你定义的BigQuery表Schema（无论是手动指定还是自动推断），确保每个字段的类型（如STRING/INTEGER/FLOAT）和本地JSON数据中对应字段的实际类型完全一致。
• 典型错误场景：BigQuery中某字段定义为STRING，但JSON里该字段的值是整数（比如{"user_id": 123}而非{"user_id": "123"}），Beam在处理字符串类型字段时会尝试调用encode，直接传入int就会触发报错。

2. 检查数据预处理逻辑 #

• 如果你在读取JSON后做了数据转换操作，确认没有误改字段类型：比如把原本应为字符串的字段转成了整数，或者遗漏了将整数转成字符串的步骤。
• 可在WriteToBigQuery步骤前添加日志输出，打印每条数据的字段类型，快速定位异常字段：

  def log_data_types(element):
      type_info = {k: str(type(v)) for k, v in element.items()}
      print(f"Data: {element}, Type Info: {type_info}")
      return element
  
  pipeline | "Log Data Types" >> beam.Map(log_data_types)
  

3. 验证WriteToBigQuery的配置参数 #

• 若你手动指定了schema参数，确认Schema格式正确：比如使用BigQuery Schema字符串（如"user_id:INTEGER,user_name:STRING"）或JSON Schema时，字段类型没有写错。
• 若开启了自动创建表（create_disposition=CreateDisposition.CREATE_IF_NEEDED），Beam可能会错误推断字段类型（比如把整数推断为字符串），此时建议手动指定准确的Schema。

4. 定位报错源码的逻辑 #

报错堆栈指向bigquery.py:2157的lambda函数，这个lambda通常是在为BigQuery写入做字段值编码处理。当字段被标记为STRING类型时，Beam会尝试对值进行编码，但如果值是int类型，直接调用encode就会失败。你可以：

• 查看该lambda的处理逻辑，确认它针对的是哪个字段；
• 验证该字段在数据中的实际类型是否与Schema定义匹配。

5. 单条数据测试 #

提取一条触发报错的JSON数据，构造最小测试用例单独写入BigQuery，快速定位问题字段。比如：

import apache_beam as beam
from apache_beam.io.gcp.bigquery import WriteToBigQuery

test_data = [{"user_id": 123, "user_name": "test"}]

with beam.Pipeline() as p:
    (p | beam.Create(test_data)
       | WriteToBigQuery(
           table="project.dataset.table",
           schema="user_id:STRING,user_name:STRING",  # 故意设置错误类型模拟报错
           write_disposition="WRITE_TRUNCATE"
       ))

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