Flink字符串转JSON的坑与优化方法

本文深入剖析了在 Apache Flink 中将 JSON 字符串转换为 JSONObject 时遇到的常见问题，特别是因不当的 JSON 库使用及对象实例化方式导致的 `NullPointerException`。通过示例代码展示了错误的使用方式，并详细阐述了如何通过正确使用 `org.json` 库的 `JSONObject` 构造函数来解决此问题，确保 Flink 作业的稳定运行。此外，文章还强调了直接在 Flink 应用中传递 `JSONObject` 的性能劣势，推荐使用 POJO (Plain Old Java Object) 进行高效序列化和反序列化的最佳实践，从而提升 Flink 流处理应用的性能和类型安全性。掌握这些技巧，能有效避免 Flink JSON 处理中的常见坑，提升应用质量。

1. 问题背景：Flink 中 JSON 字符串转换的挑战

在 Apache Flink 流处理应用中，经常需要从数据源（如 Kafka、文件等）接收 JSON 格式的字符串数据，并将其解析为可操作的 JSONObject 对象。然而，这一看似简单的转换过程有时会遇到意想不到的错误。一个常见的场景是，当尝试将解析后的 JSONObject 通过 Collector 发送时，Flink 作业会抛出 java.lang.NullPointerException: Assigned key must not be null! 异常，导致作业失败。尽管调试显示字符串已成功解析为 JSONObject 实例，但 Flink 的内部机制却无法正确处理这些对象。

2. 错误示例与问题分析

让我们首先审视导致此问题的典型代码结构。以下是一个尝试在 Flink ProcessFunction 中将 String 转换为 JSONObject 的示例：

import com.alibaba.fastjson.JSONObject; // 假设使用了FastJSON库的JSONObject

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

public class FlinkJsonProcessingExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputDS = env.fromElements(
            "{\"bill_info\":{\"ADD_TIME\":\"2022-11-12 16:05:28:418\",\"ORDER_ID\":\"f3e60c5f-78f6-4ec6-bab6-98b177f7cb67\",\"ADDER_NO\":\"0706\",\"UPDATER_NO\":\"0706\",\"S_USER_ID\":\"s68\",\"B_USER_ID\":\"b77\",\"BILL_ID\":\"8687b584-038c-498c-8f97-ec1ca197da96\",\"ADDER_NAME\":\"sss\",\"UPDATE_TIME\":\"2022-11-12 16:05:28:418\",\"UPDATER_NAME\":\"ssss\"}}"
        );

        SingleOutputStreamOperator<JSONObject> jsonObjDS = inputDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                // 问题所在：可能使用了不兼容的JSONObject解析方法
                JSONObject jsonObject = JSONObject.parseObject(value); // 假设这里使用了FastJSON的parseObject
                out.collect(jsonObject); // 在此处抛出异常
            }
        });
        jsonObjDS.print();

        env.execute();
    }
}

运行上述代码时，可能会遇到以下异常信息：

Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
Caused by: java.lang.RuntimeException: Assigned key must not be null!
Caused by: java.lang.NullPointerException: Assigned key must not be null!

这个 NullPointerException: Assigned key must not be null! 异常通常与 Flink 内部的状态管理或数据序列化机制有关。尽管 JSONObject.parseObject(value) 可能成功将字符串转换为 JSONObject 对象，但如果这个 JSONObject 对象（例如，来自 com.alibaba.fastjson 库）与 Flink 内部默认期望的 JSONObject 类型（例如，来自 org.json 库）不兼容，或者其内部结构不符合 Flink 序列化器的预期，就可能导致在 out.collect() 时出现问题。

3. 正确的转换方法与代码实现

解决这个问题的关键在于确保 JSONObject 对象的正确实例化，并与 Flink 环境中使用的 JSON 库保持一致。如果目标是使用 org.json 库提供的 JSONObject 类型，那么正确的做法是利用其构造函数来解析 JSON 字符串。

3.1 引入正确的依赖

首先，请确保您的 pom.xml 文件中包含了 org.json 库的依赖：

<dependency>
    <groupId>org.json</groupId>
    <artifactId>json</artifactId>
    <version>20180130</version> <!-- 或更高版本 -->
</dependency>

3.2 修正代码逻辑

将 ProcessFunction 中的 JSONObject 实例化方式从 JSONObject.parseObject(value) 修改为 new JSONObject(value)：

import org.json.JSONObject; // 确保导入的是org.json库的JSONObject

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

public class FlinkJsonProcessingSolution {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputDS = env.fromElements(
            "{\"bill_info\":{\"ADD_TIME\":\"2022-11-12 16:05:28:418\",\"ORDER_ID\":\"f3e60c5f-78f6-4ec6-bab6-98b177f7cb67\",\"ADDER_NO\":\"0706\",\"UPDATER_NO\":\"0706\",\"S_USER_ID\":\"s68\",\"B_USER_ID\":\"b77\",\"BILL_ID\":\"8687b584-038c-498c-8f97-ec1ca197da96\",\"ADDER_NAME\":\"sss\",\"UPDATE_TIME\":\"2022-11-12 16:05:28:418\",\"UPDATER_NAME\":\"ssss\"}}"
        );

        SingleOutputStreamOperator<JSONObject> jsonObjDS = inputDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                // 正确的做法：使用org.json.JSONObject的构造函数
                JSONObject jsonObject = new JSONObject(value);
                out.collect(jsonObject);
            }
        });
        jsonObjDS.print();

        env.execute();
    }
}

通过这一修改，Flink 作业将能够成功运行并打印出解析后的 JSONObject，例如：

{"bill_info":{"ADDER_NAME":"sss","ADDER_NO":"0706","UPDATER_NAME":"ssss","UPDATER_NO":"0706","BILL_ID":"8687b584-038c-498c-8f97-ec1ca197da96","ADD_TIME":"2022-11-12 16:05:28:418","ORDER_ID":"f3e60c5f-78f6-4ec6-bab6-98b177f7cb67","S_USER_ID":"s68","B_USER_ID":"b77","UPDATE_TIME":"2022-11-12 16:05:28:418"}}

这个解决方案的关键在于，org.json.JSONObject 的构造函数 new JSONObject(String source) 能够直接从 JSON 字符串构建一个 JSONObject 实例，并且这个实例与 Flink 内部默认的序列化器对 org.json.JSONObject 的处理方式是兼容的。而如果混合使用不同 JSON 库的 API（例如，使用 com.alibaba.fastjson.JSONObject.parseObject 来创建对象，但 Flink 的类型信息或序列化器期望的是 org.json.JSONObject），就可能导致上述的 NullPointerException。

4. 关键点与注意事项

4.1 JSON 库的选择与统一

Java 生态系统中有多种 JSON 处理库，如 org.json、FastJSON、Jackson、Gson 等。在 Flink 应用中，选择一个库并保持其在整个项目中的一致性至关重要。如果您的项目依赖了多个 JSON 库，请务必明确您正在使用的 JSONObject 类是来自哪个库，并使用该库提供的正确 API 进行解析和操作。本教程中的解决方案是针对 org.json 库的。

4.2 依赖管理

确保 Maven 或 Gradle 依赖中只包含您实际使用的 JSON 库，并避免版本冲突。不必要的依赖或冲突可能导致类加载问题，进而引发运行时错误。

4.3 性能考量与最佳实践：优先使用 POJO

尽管将 JSON 字符串直接转换为 JSONObject 看起来很方便，但在 Flink 这样的高性能流处理框架中，直接在数据流中传递 JSONObject 通常不是最佳实践。原因如下：

• 序列化/反序列化开销大： JSONObject 是一种通用容器，其内部结构复杂，字段类型不固定。这使得 Flink 在序列化和反序列化这些对象时需要进行更多的反射和类型推断，导致显著的性能开销。
• 缺乏类型安全： 直接操作 JSONObject 需要通过字符串键来访问字段，容易出错且缺乏编译时检查。
• 代码可读性与维护性： 使用 POJO 可以使代码更清晰、更易读，因为它明确定义了数据的结构。

推荐的做法是： 将 JSON 字符串反序列化为 POJO（Plain Old Java Object）。POJO 具有明确的结构和类型，Fink 可以对其进行高效的序列化（如 Kryo 序列化器），从而显著提升性能并增强类型安全性。

示例：将 JSON 转换为 POJO

1. 定义 POJO 类：

   public class BillInfo {
       public String ADD_TIME;
       public String ORDER_ID;
       public String ADDER_NO;
       public String UPDATER_NO;
       public String S_USER_ID;
       public String B_USER_ID;
       public String BILL_ID;
       public String ADDER_NAME;
       public String UPDATE_TIME;
       public String UPDATER_NAME;
   
       // 默认构造函数是必需的
       public BillInfo() {}
   
       // 可以在这里添加getter/setter或其他业务逻辑
   }
   
   public class BillData {
       public BillInfo bill_info;
   
       public BillData() {}
   }

2. 使用 MapFunction 或 ProcessFunction 结合 JSON 库（如 Jackson 或 Gson）进行反序列化：

   import com.fasterxml.jackson.databind.ObjectMapper; // 示例使用Jackson
   
   // ... Flink setup code ...
   
   SingleOutputStreamOperator<BillData> billDataDS = inputDS.map(jsonString -> {
       ObjectMapper mapper = new ObjectMapper();
       return mapper.readValue(jsonString, BillData.class);
   });
   
   billDataDS.print();
   env.execute();

   或者，对于更复杂的场景，可以实现一个自定义的 DeserializationSchema。

5. 总结

在 Flink 中处理 JSON 字符串时，正确选择和使用 JSON 库的 API 是避免 NullPointerException 的关键。特别是当使用 org.json 库时，应通过 new JSONObject(String source) 构造函数来实例化对象。然而，从性能和类型安全的角度考虑，将 JSON 字符串反序列化为 POJO 是更推荐的实践。通过采用 POJO，您可以构建更健壮、高效且易于维护的 Flink 流处理应用。

到这里，我们也就讲完了《Flink字符串转JSON的坑与优化方法》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！