如何定期从数据库刷新Apache Beam侧输入？报错求助

原代码问题分析 #

• 误用JdbcIO.readAll()：该API会为输入的每个元素执行一次数据库查询，而你的场景是定期执行一次全量查询，用它会导致不必要的重复查询，且容易引发链式操作异常。
• 窗口逻辑冗余：先做FixedWindow再转GlobalWindow，增加了处理复杂度，且触发器配置和窗口的组合可能导致元素流转异常。

正确实现方式 #

核心思路是：用定时触发信号驱动ParDo执行数据库全量查询，通过GlobalWindow+触发器控制刷新频率，最终将结果转为可刷新的侧输入。

1. 实现定期读取数据库的ParDo #

import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.KV;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;

public class ReadUserDbDoFn extends DoFn<Long, KV<String, String>> {
    private final SerializableFunction<Void, DataSource> dataSourceProviderFn;

    public ReadUserDbDoFn(SerializableFunction<Void, DataSource> dataSourceProviderFn) {
        this.dataSourceProviderFn = dataSourceProviderFn;
    }

    @ProcessElement
    public void processElement(ProcessContext ctx) throws Exception {
        // 每次触发时全量读取数据库
        try (Connection conn = dataSourceProviderFn.apply(null).getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery("select id, concat(first_name, ' ', last_name) from users")) {
            while (rs.next()) {
                ctx.output(KV.of(rs.getString(1), rs.getString(2)));
            }
        }
    }
}

2. 构建可定期刷新的侧输入 #

final PCollectionView<Map<String, String>> userMap = pipeline
        // 每120秒生成一个触发信号
        .apply("Generate Refresh Signal", GenerateSequence.from(0)
                .withRate(1, Duration.standardSeconds(120L)))
        // 绑定到GlobalWindow，配合触发器控制刷新时机
        .apply("Configure Global Window", Window.<Long>into(new GlobalWindows())
                .triggering(Repeatedly.forever(
                        AfterProcessingTime.pastFirstElementInPane()
                                .plusDelayOf(Duration.standardSeconds(0))))
                .discardingFiredPanes()
                .withAllowedLateness(Duration.ZERO))
        // 执行数据库读取
        .apply("Read User Data", ParDo.of(new ReadUserDbDoFn(userDsFn)))
        // 转为Map类型的侧输入，每次触发都会更新内容
        .apply("As Map View", View.<String, String>asMap());

3. 侧输入的使用示例 #

在主数据流的ParDo中引用该侧输入，每次处理主数据时都会获取最新的用户映射：

mainData.apply("Process With Fresh Side Input", ParDo.of(new DoFn<MainData, Result>() {
    @ProcessElement
    public void processElement(ProcessContext ctx) {
        Map<String, String> latestUserMap = ctx.sideInput(userMap);
        // 使用最新的用户映射处理主数据
        MainData data = ctx.element();
        String userName = latestUserMap.get(data.getUserId());
        // ... 后续处理逻辑
    }
}).withSideInputs(userMap));

关键说明 #

• 触发器配置：AfterProcessingTime.pastFirstElementInPane()确保每个触发信号到达后立即执行查询，Repeatedly.forever()保证持续定期触发。
• discardingFiredPanes：丢弃已触发的窗口数据，避免侧输入累积旧数据，确保每次都是最新的查询结果。
• ParDo读取数据库：相比JdbcIO.readAll，ParDo更适合这种定期全量读取的场景，能更灵活控制查询时机和逻辑，避免API误用带来的异常。

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