Apache Beam使用WriteToJdbc写入MySQL时遇Unknown Coder URN及Schema问题求助

看起来你遇到的这两个错误都是因为WriteToJdbc是跨语言实现的组件（底层依赖Java的Expansion Service）导致的：它不支持Python默认的pickle编码器，同时要求输入的PCollection必须带有明确的Schema，和目标MySQL表的结构匹配。下面给你一步步的解决思路和修改后的代码：

问题根源拆解 #

1. Unknown Coder URN错误：Python默认用pickle序列化数据，但Java端的Expansion Service不识别这个编码器的URN，必须使用Beam跨语言支持的标准编码器（比如RowCoder）。
2. Cannot call getSchema错误：WriteToJdbc需要明确的Schema来映射MySQL表的字段，你之前指定typing.Iterable[str]是无效的，因为它不是结构化的行数据Schema。

解决方案步骤 #

1. 定义与MySQL表匹配的Schema #

首先你需要根据目标MySQL表的字段，定义对应的结构化Schema。这里推荐用NamedTuple，更直观易懂。

比如假设你的MySQL表有id(INT)、username(VARCHAR)、email(VARCHAR)、age(INT, 允许为空)这几个字段，你可以这样定义：

from typing import NamedTuple, Optional
import apache_beam.coders as coders

# 定义和MySQL表字段完全匹配的NamedTuple
class DbRow(NamedTuple):
    id: int
    username: str
    email: str
    age: Optional[int]  # 允许为空的字段用Optional标注

# 注册对应的RowCoder，让Beam能正确序列化这个类型
coders.registry.register_coder(DbRow, coders.RowCoder)

2. 修改数据处理逻辑，输出符合Schema的对象 #

在ProcessData的process方法里，需要把flatten后的JSON数据转换为刚才定义的DbRow对象，同时处理JSON中可能缺失的字段（用默认值或者None，要和MySQL表的字段约束一致）：

class ProcessData(beam.DoFn):
    def process(self, element):
        element = json.loads(element)
        flat_json = flatsplode.flatsplode(element, "_")
        for data in flat_json:
            # 转换为DbRow对象，处理缺失字段的情况
            yield DbRow(
                id=data.get("id", 0),  # 如果id缺失，用默认值0（可根据表约束调整）
                username=data.get("username", ""),
                email=data.get("email", ""),
                age=data.get("age")  # 缺失的话会是None，对应MySQL的NULL
            )

3. 调整Pipeline的写入逻辑 #

现在你的process_data PCollection已经是带有明确Schema的DbRow对象了，直接传给WriteToJdbc即可，不需要额外的类型标注：

process_data | f"Write to RDBMS" >> WriteToJdbc(
    table_name=data_targets["rdbms"]["tablename"],
    driver_class_name=data_targets["rdbms"]['driver_class_name'],
    jdbc_url=data_targets["rdbms"]['jdbc_url'],
    username=data_targets["rdbms"]['username'],
    password=data_targets["rdbms"]['password']
)

为什么你之前的尝试失败？ #

• 用typing.Iterable[str]：这个类型是字符串列表，完全不符合WriteToJdbc需要的结构化行数据要求，所以会触发Schema缺失的错误。
• 注册NamedTuple时的pcoll_schema：你可能没有正确定义这个Schema，导致Beam无法识别对应的结构，必须确保NamedTuple的字段和MySQL表完全匹配，并且正确注册RowCoder。

额外注意事项 #

• 如果你的JSON字段和MySQL表字段名称不一致，可以在转换的时候做映射（比如JSON里的user_id对应表的id）。
• 确保MySQL表的字段类型和你定义的NamedTuple类型兼容（比如Python的int对应MySQL的INT，str对应VARCHAR等）。
• 如果你的记录确实有动态字段，可能需要先做字段过滤，只保留MySQL表中存在的字段，否则WriteToJdbc会因为字段不匹配报错。

备注：内容来源于stack exchange，提问作者somnath chouwdhury