使用Apache Beam Python SDK ReadFromJdbc读Postgres报javasdk逻辑类型错误

根因说明 #

ValueError: No logical type registered for URN 'beam:logical_type:javasdk:v1' 错误产生的原因是Beam的JDBC IO属于跨语言实现，依赖Java侧的服务逻辑，Python侧默认没有注册对应Java SDK的逻辑类型映射，或是Python、Java两侧的Beam SDK版本不兼容导致。

排查步骤 #

• 核对Apache Beam Python SDK与Java SDK的版本号是否完全一致，跨语言IO要求两侧版本严格匹配，不可出现版本号错位的情况。Python侧版本可通过pip show apache-beam命令查看，Java侧版本可核对你安装的Beam Java SDK包版本。
• 检查是否显式触发了Java逻辑类型的注册逻辑，该注册逻辑默认不会被自动导入。
• 检查是否配置了正确的PostgreSQL JDBC驱动路径，避免后续出现驱动类找不到的问题。

具体修复操作 #

1. 对齐两侧SDK版本后，在代码导入区域新增Java逻辑类型注册相关代码：

from apache_beam.typehints import logical_types
# 显式注册Java SDK逻辑类型
logical_types.register_java_logical_type()

2. 实例化PipelineOptions时添加跨语言运行所需的实验参数：

beam_options = PipelineOptions(
    experiments=['use_runner_v2']
)

3. 在ReadFromJdbc参数中补充PostgreSQL JDBC驱动包路径，提前下载对应版本的PostgreSQL JDBC jar包到本地后添加配置：

| 'Read from jdbc' >> ReadFromJdbc(
    table_name='jdbc_external_test_read',
    driver_class_name='org.postgresql.Driver',
    jdbc_url='jdbc:postgresql://localhost:5432/example',
    username='postgres',
    password='postgres',
    jars=['./postgresql-42.6.0.jar'] # 替换为你本地实际的jar包路径
)

修改后完整可运行代码示例 #

import logging
import typing

import apache_beam as beam
from apache_beam import coders
from apache_beam.io.jdbc import ReadFromJdbc
from apache_beam.options.pipeline_options import PipelineOptions
from past.builtins import unicode
# 新增导入
from apache_beam.typehints import logical_types
logical_types.register_java_logical_type()


def run(argv=None):
    # 新增实验参数
    beam_options = PipelineOptions(
        experiments=['use_runner_v2']
    )

    ExampleRow = typing.NamedTuple('ExampleRow', [('id', int), ('name', unicode)])

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

    with beam.Pipeline(options=beam_options) as p:
        result = (
            p
            | 'Read from jdbc' >> ReadFromJdbc(
                                    table_name='jdbc_external_test_read',
                                    driver_class_name='org.postgresql.Driver',
                                    jdbc_url='jdbc:postgresql://localhost:5432/example',
                                    username='postgres',
                                    password='postgres',
                                    # 替换为本地实际的JDBC驱动jar包路径
                                    jars=['./postgresql-42.6.0.jar']
                                )
            | 'Print result' >> beam.Map(print)
        )


if __name__ == '__main__':
    logging.getLogger(__name__).setLevel(logging.INFO)
    run()

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