Apache Beam Python SDK读取PostgreSQL时timestamp类型解析报错求助

环境信息 #

• Python v3.11.4
• apache-beam v2.51.0（Python SDK）
• postgres v11.5
• DirectRunner

问题描述 #

使用apache_beam.io.jdbc组件读写PostgreSQL时，整数、字符串等基本数据类型处理正常，但解析timestamp without time zone类型字段时触发错误，报错指向MicrosInstant逻辑类型解析失败，已确认时间戳字段无NULL值，同时伴随Java侧的连接警告。

管道代码示例 #

with beam.Pipeline(options=None) as p:
    pipeline = (
        p
        | ReadFromJdbc(
            table_name="table_name",
            driver_class_name='org.postgresql.Driver',
            jdbc_url='jdbc:{}://{}:{}/{}'.format("postgresql", "127.0.0.1", "5432", "db_name"),
            username="postgres",
            password="redacted",
            query="SELECT * FROM table_name")
        | beam.Map(print)
    )

错误信息 #

Python解析错误栈 #

File "apache_beam/coders/coder_impl.py", line 1890, in apache_beam.coders.coder_impl.LogicalTypeCoderImpl.decode_from_stream
  File "/Users/archit.shah/PycharmProjects/duplopy-pysql-beam/venv-3.11/lib/python3.11/site-packages/apache_beam/typehints/schemas.py", line 873, in to_language_type
    return Timestamp(seconds=int(value.seconds), micros=int(value.micros))
                             ^^^^^^^^^^^^^^^^^^
TypeError: int() argument must be a string, a bytes-like object or a real number, not 'NoneType'

Java警告信息 #

WARNING:root:severity: WARN
timestamp {
  seconds: 1697642143
  nanos: 162000000
}
message: "Hanged up for url: \"host.docker.internal:58970\"\n."
log_location: "org.apache.beam.sdk.fn.data.BeamFnDataGrpcMultiplexer"
thread: "16"

解决思路建议 #

1. 显式转换SQL查询中的时间戳字段 #

在查询语句中将timestamp without time zone字段转换为字符串或epoch秒数，绕过Beam默认的逻辑类型解析：

SELECT id, name, created_at::text AS created_at FROM table_name

或者转换为数值类型：

SELECT id, name, EXTRACT(EPOCH FROM created_at)::bigint AS created_at FROM table_name

之后在Python侧将结果转换为datetime对象处理即可。

2. 自定义编码器处理时间戳类型 #

注册自定义Coder替代默认的MicrosInstant编码器，直接处理时间戳的编解码：

from apache_beam.coders import Coder
import datetime

class TimestampCoder(Coder):
    def encode(self, value):
        return value.isoformat().encode('utf-8')

    def decode(self, encoded):
        return datetime.datetime.fromisoformat(encoded.decode('utf-8'))

# 在管道中应用自定义编码器处理目标字段
with beam.Pipeline(options=None) as p:
    pipeline = (
        p
        | ReadFromJdbc(
            table_name="table_name",
            driver_class_name='org.postgresql.Driver',
            jdbc_url='jdbc:postgresql://127.0.0.1:5432/db_name',
            username="postgres",
            password="redacted",
            query="SELECT * FROM table_name")
        | beam.Map(lambda row: row._replace(created_at=TimestampCoder().decode(row.created_at)))
        | beam.Map(print)
    )

注意：需根据实际表结构中的字段名调整替换逻辑。

3. 检查JDBC驱动兼容性 #

当前使用的PostgreSQL JDBC驱动可能与beam v2.51.0存在兼容性问题，尝试升级驱动到最新稳定版（如42.6.0），确保启动Beam时能加载到正确的驱动文件。

4. 显式定义输出Schema #

通过RowSchema提前定义字段类型，指定时间戳字段为datetime类型，引导Beam按指定类型解析：

from apache_beam.typehints.schemas import RowSchema
from apache_beam.typehints import datetime as beam_datetime

# 根据实际表结构定义Schema
schema = RowSchema.from_fields([
    ('id', int),
    ('name', str),
    ('created_at', beam_datetime)
])

with beam.Pipeline(options=None) as p:
    pipeline = (
        p
        | ReadFromJdbc(
            table_name="table_name",
            driver_class_name='org.postgresql.Driver',
            jdbc_url='jdbc:postgresql://127.0.0.1:5432/db_name',
            username="postgres",
            password="redacted",
            query="SELECT * FROM table_name",
            output_schema=schema)
        | beam.Map(print)
    )

5. 替代方案参考 #

如果apache_beam.io.jdbc的问题无法快速解决，可以尝试：

• 使用PostgreSQL导出Parquet文件，再通过apache_beam.io.parquet读取；
• 在DoFn中直接使用psycopg2处理数据库连接（注意分布式运行时的连接池控制）。

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