4. Flink 环境搭建：本地开发环境、Flink 集群模式、Kafka/Pulsar 数据源配置

在商户风险迁移场景中，批次偏移诊断依赖稳定、可复现的 Flink 运行环境。本章从本地开发环境、集群部署模式、消息队列数据源配置三个维度展开，确保你能够快速搭建一套可用于“偏移量回溯”与“风险批次重算”的 Flink 实验平台。

4.1 本地开发环境搭建

本地环境主要用于代码调试、单元测试与偏移量逻辑验证。推荐使用 Flink 1.17+ 配合 Java 11 或 Java 17，并启用 Checkpointing 模拟状态恢复。

4.2 Flink 集群模式对比

生产环境中的商户风险迁移任务通常运行在 YARN 或 Kubernetes 上。下表对比三种常见部署模式在偏移量诊断场景下的适用性：

4.3 Kafka / Pulsar 数据源配置

商户风险迁移的输入流通常来自 Kafka 或 Pulsar，其中包含商户交易、风险评分、批次状态变更等事件。偏移量诊断的核心在于精确记录每个分区的消费位置。

4.3.1 Kafka Source 配置要点

// Flink 1.17 Kafka Source 示例
DataStream<String> riskStream = KafkaSource.<String>builder()
    .setBootstrapServers("kafka1:9092,kafka2:9092")
    .setTopics("merchant-risk-events")
    .setGroupId("risk-migration-group")
    .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .setProperties(immutableMapOf(
        "enable.auto.commit", "false",
        "partition.discovery.interval.ms", "30000"
    ))
    .build();

env.fromSource(riskStream, WatermarkStrategy.noWatermarks(), "Kafka Risk Source");

• committedOffsets + EARLIEST：优先从上次提交偏移量开始消费，无提交记录则从最早位置开始，适合偏移量回溯。
• enable.auto.commit=false：由 Flink Checkpoint 控制偏移量提交，避免重复消费或丢失。
• partition.discovery.interval.ms：动态发现新增分区，防止风险批次遗漏。

4.3.2 Pulsar Source 配置要点

// Flink 1.17 Pulsar Source 示例
DataStream<String> pulsarStream = PulsarSource.<String>builder()
    .setServiceUrl("pulsar://pulsar1:6650")
    .setAdminUrl("http://pulsar1:8080")
    .setTopics("persistent://public/default/risk-batch")
    .setSubscriptionName("risk-migration-sub")
    .setSubscriptionType(SubscriptionType.Exclusive)
    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
    .setConfig(immutableMapOf(
        "pulsar.source.enableAutoCommit", "false",
        "pulsar.source.startFromMessageId", "latest"
    ))
    .build();

env.fromSource(pulsarStream, WatermarkStrategy.noWatermarks(), "Pulsar Risk Source");

• Exclusive 订阅：确保同一分区只被一个 Reader 消费，避免偏移量冲突。
• enableAutoCommit=false：与 Kafka 同理，由 Flink Checkpoint 统一管理偏移量。
• startFromMessageId=latest：生产环境建议从最新消息开始，避免重放历史数据造成压力。

4.4 环境架构总览

下图展示了本地开发环境与集群模式下，Flink 如何连接 Kafka/Pulsar 数据源，并通过 Checkpoint 机制管理偏移量：

4.5 环境验证与偏移量测试

搭建完成后，建议执行以下验证步骤，确保偏移量诊断链路可用：

1. 启动 Flink 集群：本地执行 ./bin/start-cluster.sh，访问 http://localhost:8081 确认 Web UI 正常。
2. 创建测试 Topic：在 Kafka 中创建 merchant-risk-events 分区数为 3，在 Pulsar 中创建 risk-batch 分区数为 3。
3. 提交偏移量诊断 Job：使用 flink run -m yarn-cluster -yn 2 提交包含 KafkaSource 或 PulsarSource 的 JAR。
4. 模拟偏移量异常：手动停止 Job，修改 Kafka 或 Pulsar 的偏移量（如重置到 2 小时前），然后从 Savepoint 恢复，观察是否按预期回溯。
5. 验证 Checkpoint 偏移量：在 Flink Web UI 的 Checkpoints 页面查看 KafkaSourceReader 或 PulsarSourceReader 的偏移量提交记录。

4.6 常见问题与避坑