Apache Beam中JdbcIO.read重构为DoFn后无返回结果问题

• 核心错误：Beam的Transform是声明式的，必须挂载到Pipeline的执行图上才会运行。你在DoFn的方法里只是构造了JdbcIO.read的配置对象，没有调用apply将它加入Pipeline执行链，所以这部分代码完全不会被执行，自然不会触发查询和RowMapper逻辑。
• 写法错误：JdbcIO的RowMapper本身已经封装了ResultSet的遍历逻辑，Beam会自动遍历每一行数据，每一行调用一次mapRow方法，你自己在mapRow里写while(resultSet.next())属于多余逻辑，就算IO能正常运行也会导致行读取异常。
• 不符合Beam编程模型：DoFn是处理每个输入元素的计算逻辑，不支持在DoFn内部嵌套定义其他IO Transform，IO Transform必须直接作为Pipeline的apply参数，和你最初的可运行写法逻辑一致。

如果你要把JdbcIO读取逻辑封装成独立的模块，可以直接封装成返回PTransform的类，而不是继承DoFn，示例如下：

// 封装独立的Jdbc读取PTransform
public class ReadInputOtc extends PTransform<PBegin, PCollection<TestRow>> {
    private final DataSource dataSource;
    private final String query;

    public ReadInputOtc(DataSource dataSource, String query) {
        this.dataSource = dataSource;
        this.query = query;
    }

    @Override
    public PCollection<TestRow> expand(PBegin input) {
        return input.apply(JdbcIO.<TestRow>read()
                .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(dataSource))
                .withCoder(SerializableCoder.of(TestRow.class))
                .withQuery(query)
                .withRowMapper(resultSet -> {
                    TestRow testRow = new TestRow();
                    // 直接读取当前行字段赋值，不需要手动调用next()
                    testRow.setId(resultSet.getString("id"));
                    // 其他字段赋值逻辑
                    return testRow;
                }));
    }
}

调用方式和你最初的写法保持一致：

Pipeline p = createPipeline(options);
p.apply(new ReadInputOtc(dataSource, "你的查询语句"))
 .apply(MapElements.via(new SimpleFunction<TestRow, String>() {
     @Override
     public String apply(TestRow input) {
         return input.toString();
     }
 }));
// 最后执行Pipeline
p.run();

如果你的业务场景确实需要基于上游的动态参数触发查询（比如上游是查询参数集合），则不要使用JdbcIO，直接在DoFn里自己编写原生JDBC连接、查询、遍历结果集的逻辑即可，不要嵌套Beam的IO Transform。

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