Flink批处理作业无报错提前结束，仅处理部分数据如何排查

1 优先修复代码已知缺陷 #

你提供的代码存在线程安全问题，是数据丢失的高概率原因：

• Jsonizer类中logObject被定义为静态成员变量，Flink算子的多并行子任务会共享该静态变量，并发解析时会互相覆盖实例，导致大量记录被丢弃。修改方案：将logObject改为map方法内的局部变量：

public RegulatedZeekConnRecord map(String record) {
    // Initialize gson with customized deserializer
    if (gson == null) {
        gb.registerTypeAdapter(RegulatedZeekConnRecord.class, new ConnLogDeserializer());
        gson = gb.create();
    }
    // 改为方法局部变量，避免线程安全问题
    RegulatedZeekConnRecord logObject = gson.fromJson(record, RegulatedZeekConnRecord.class);
    return logObject;
}

• 检查RegulatedZeekConnRecord构造函数、ConnLogDeserializer反序列化逻辑，是否存在捕获异常后静默返回null的逻辑，这类逻辑会导致解析失败的记录被直接过滤，不会触发作业报错。

2 分阶段定位丢数环节 #

通过统计各算子的实际处理记录数，快速定位丢数位置：

• 临时简化作业逻辑：去掉groupBy、reduce逻辑，直接将读取到的原始日志行写入输出文件，统计输出行数是否和原文件22万条一致。如果一致说明问题出在聚合逻辑，否则问题出在读取/解析阶段。
• 为各算子添加Flink累加器统计记录数：分别统计Source读取总条数、解析成功条数、解析失败条数、聚合后条数，直接对比数值就能定位丢数的具体阶段。

3 运行日志与Dashboard排查 #

• 查看TaskManager、JobManager的WARN级别日志，非致命错误（如单条记录解析失败、文件分片读取异常）不会导致作业失败，只会打印WARN日志。
• 核对Dashboard中每个算子的输入输出指标：
  • 如果Source端输出记录数就只有3万，说明是文件读取问题，需要检查输入文件是否所有节点都可读、文件分片是否正常、文件路径配置是否正确。
  • 如果Source端输出22万，Jsonizer Map算子输出只有3万，说明是解析阶段丢数，重点排查反序列化逻辑。

4 数据与配置验证 #

• 检查输入文件格式：确认是否每行一条标准JSON，是否存在空行、多行JSON、格式异常的行，这类异常数据如果未做异常处理会直接丢失。
• 确认运行模式：如果是集群模式运行，使用本地文件路径需要保证所有TaskManager节点的对应路径下都有完整的输入文件，否则只会读取到部分节点上的分片数据。

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