Hazelcast Jet CDC流源转批量处理及空数据问题咨询

问题根因 #

读取Map为空的核心原因是Postgres CDC属于无限流源，默认情况下写入Map的流作业会持续运行不会主动终止，你启动读Map的批量作业时，CDC的全量快照数据还未完成写入，甚至还没开始写入，自然返回空结果。

你要实现CDC流源转批量处理的核心逻辑是：先等CDC完成全量表数据的快照同步，再触发后续的批量处理逻辑，不需要监听后续的增量变更。

可行实现方案 #

方案1：等快照完成终止流作业，再启动批量处理 #

Hazelcast Postgres CDC基于Debezium实现，全量快照阶段的ChangeRecord会携带snapshot标识，所有快照数据推送完成后，你可以主动终止写入Map的作业，再启动批量读Map的作业即可，示例代码如下：

// 初始化分布式标记，用于判断快照是否完成
IAtomicLong snapshotFlag = hazelcastInstance.getCPSubsystem().getAtomicLong("snapshotFinishFlag");
snapshotFlag.set(0);

// 构造写Map的流作业
Pipeline writePipe = Pipeline.create();
writePipe.readFrom(source)
        .withoutTimestamps()
        .writeTo(Sinks.map("dbStreamedData", ChangeRecord::key, ChangeRecord::value));

Job writeJob = hazelcastInstance.newJob(writePipe);

// 异步监听快照完成状态
new Thread(() -> {
    // 生产环境建议通过记录的snapshot标识判断：当连续收到的记录__dbz_snapshot头从true变为false时，说明快照完成
    // 此处为简化示例，根据你的表数据量调整等待时长
    try {
        Thread.sleep(300000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    // 快照完成后终止流作业
    writeJob.cancel();
    snapshotFlag.set(1);
}).start();

// 阻塞等待快照完成
while (snapshotFlag.get() != 1) {
    Thread.sleep(1000);
}

// 此时Map已写入全量数据，启动批量处理作业
Pipeline batchPipe = Pipeline.create();
batchPipe.readFrom(Sources.map("dbStreamedData"))
        // 此处插入你依赖用户元数据的批量处理逻辑
        .writeTo(Sinks.logger());
hazelcastInstance.newJob(batchPipe).join();

方案2：同一Pipeline内用窗口实现流转批量 #

如果不想拆分两个作业，可以用全局窗口把所有快照数据攒成一个批次处理：

Pipeline pipeline = Pipeline.create();
pipeline.readFrom(source)
        .withoutTimestamps()
        // 定义全局窗口，触发条件为快照阶段结束
        .window(WindowDefinition.global()
                .trigger(Trigger.onEvent((record, ts) -> !record.headers().get("__dbz_snapshot").asBoolean())))
        // 把窗口内所有数据聚合成列表，作为批量处理的输入
        .aggregate(AggregateOperations.toList())
        .map(batchRecords -> {
            // 此处插入你依赖用户元数据的批量处理逻辑
            return batchRecords;
        })
        .writeTo(Sinks.logger());

注意事项 #

• 如果业务不需要后续增量变更，快照完成后一定要终止CDC流作业，避免浪费集群资源。
• 数据量较大的场景下，中间存储的Map建议开启分区存储和持久化配置，避免单节点内存溢出。

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