Apache Beam Python使用ReadFromJDBC读Postgres字段schema解码错误如何修复

该报错出现于Apache Beam Python SDK 2.34.0及更早版本，Java端JDBC IO会将PostgreSQL的VARCHAR(n)类型封装为自定义的VariableLengthString逻辑类型，Python侧未提前注册该类型的解码器，导致schema解析失败。

你可以任选以下任一方案解决：

方案1：SQL层类型转换（最快捷） #

在查询语句中将VARCHAR类型强转为无长度限制的TEXT类型，Java端会直接将其识别为原生字符串类型，无需额外解码配置：

SELECT row_id, person_id, case_id, seq_num, code::TEXT AS code FROM table_name LIMIT 100

替换你原ReadFromJdbc配置里的query参数即可。

方案2：注册自定义逻辑类型解码器 #

如果不想修改SQL，可以在流水线启动前手动注册该逻辑类型的解码规则：

from apache_beam.typehints import schemas

# 注册Java端VariableLengthString逻辑类型的解码规则
def decode_varchar(payload, value):
    # 该逻辑类型底层存储就是标准字符串，直接返回原值即可
    return value

schemas.register_logical_type(
    urn="beam:logical_type:javasdk:v1",
    type_factory=lambda rep_type, arg: str,
    value_converter=decode_varchar
)

方案3：升级Beam版本 #

2.37.0及以上版本的Apache Beam Python SDK已经内置了JDBC相关逻辑类型的默认支持，直接将依赖升级到apache-beam>=2.37.0即可原生解决该问题。

你定义的NamedTuple需要匹配表字段的可空属性，避免空值报错：

import typing
from apache_beam import coders
from apache_beam.typehints.schemas import schema

@schema
class TableRow(typing.NamedTuple):
    row_id: int
    person_id: int
    case_id: int
    seq_num: typing.Optional[int]  # 允许为空
    code: typing.Optional[str]     # 允许为空

coders.registry.register_coder(TableRow, coders.RowCoder)

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