Apache Beam Python SDK中BigQuery TIMESTAMP转换Bug及临时解决方法咨询

当前Apache Beam Python SDK的BigQuery模块存在Bug：会将BigQuery的TIMESTAMP类型错误转换为DATETIME。该问题的修复已完成，但仅存在于预发布版本，最新稳定版2.49.0并未包含此修复。

此Bug仅在使用Storage Write API时触发输入/输出Schema不匹配错误，传统流式API可正常工作——具体表现为SDK将LOGICAL_TYPE<beam:logical_type:micros_instant:v1>转换为DATETIME而非TIMESTAMP。

临时解决方法 #

• 显式指定输出Schema：写入BigQuery时，手动定义Schema并将对应字段明确标记为TIMESTAMP，覆盖SDK的自动类型推断。示例代码：

from apache_beam.io.gcp.bigquery import BigQueryDisposition, WriteToBigQuery

# 自定义输出Schema，指定event_time为TIMESTAMP类型
output_schema = {
    "fields": [
        {"name": "event_time", "type": "TIMESTAMP", "mode": "REQUIRED"},
        # 补充其他字段定义
    ]
}

# 写入BigQuery时传入自定义Schema
pipeline | WriteToBigQuery(
    table="your-project.your-dataset.your-table",
    schema=output_schema,
    write_disposition=BigQueryDisposition.WRITE_APPEND,
    create_disposition=BigQueryDisposition.CREATE_NEVER,
    use_storage_write_api=True
)

• 转换时间字段为字符串格式：在写入前将beam:logical_type:micros_instant:v1类型的时间字段转换为BigQuery兼容的TIMESTAMP字符串格式（如YYYY-MM-DD HH:MM:SS.SSSSSS），让SDK按字符串转换后自动匹配TIMESTAMP类型。示例：

import apache_beam as beam
from datetime import datetime

class ConvertToBQTimestamp(beam.DoFn):
    def process(self, element):
        # 假设element中的event_time是微秒级时间戳
        event_dt = datetime.fromtimestamp(element["event_time"] / 1000000)
        element["event_time"] = event_dt.strftime("%Y-%m-%d %H:%M:%S.%f")
        yield element

# 在写入BigQuery前添加转换步骤
pipeline | beam.ParDo(ConvertToBQTimestamp()) | WriteToBigQuery(
    table="your-project.your-dataset.your-table",
    use_storage_write_api=True
)

• 临时降级到传统流式API：如果业务场景允许，可关闭Storage Write API，使用传统流式API绕过此问题——只需在WriteToBigQuery中设置use_storage_write_api=False。

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