实时计算 Flink版产品使用问题之如何高效地将各分片存储并跟踪每个分片的消费位置

// 假设 SourceRecord 的 sourcePartition 包含了 shardId 信息
Map<String, Object> sourcePartition = new HashMap<>();
sourcePartition.put("shardId", shardId);
// sourceOffset Map 用于存储每个分片的消费位置
Map<String, Object> sourceOffset = new HashMap<>();
// 更新 offset 时，以 shardId 作为 key，并将 pos 作为 value 存储
sourceOffset.put(shardId, pos);
// 创建或更新 SourceRecord 时，设置其 partition 和 offset 信息
SourceRecord record = new SourceRecord(
    sourcePartition,
    sourceOffset,
    topicName, // Kafka Topic 或其他目标 Sink 的名称
    schema, // 数据的 Schema
    value // CDC 记录的实际数据
);
// 在 Flink 的 SourceFunction 中，每当处理完一条记录后，需要根据新的 pos 更新 sourceOffset
public void process(SourceContext<T> context) throws Exception {
    while (true) {
        SourceRecord record = fetchNextRecord(); // 获取下一条 CDC 记录
        String shardId = (String) record.sourcePartition().get("shardId");
        Long pos = (Long) record.sourceOffset().get(shardId);
        // 处理记录
        // ...
        // 更新 offset，这里假设 fetchNextRecord 已经获取到了新的 pos
        Long newPos = getNextPosition(record); // 获取新位置
        sourceOffset.put(shardId, newPos);
        // 发送数据到 Flink 的上下文
        context.collect(transformToInternalFormat(record));
    }
}
// 确保在 checkpoint 时保存这些 offset
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
    // 使用 Flink State API 将 sourceOffset 保存至状态后端，确保在故障恢复时能够正确恢复消费位点
    stateBackend.getOrCreate(keyedOffsetsStateDescriptor).put(currentShardId, sourceOffset);
}
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
    // 从状态后端恢复 offset
    ValueState<Map<String, Object>> offsetsState = context.getKeyedStateStore().getState(keyedOffsetsStateDescriptor);
    Map<String, Object> restoredOffsets = offsetsState.get();
    if (restoredOffsets != null) {
        sourceOffset.putAll(restoredOffsets);
    }
}