Flink反序列化Kafka JSON

<dependencies>
  <!-- Flink Kafka Connector -->
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.12</artifactId>
    <version>1.14.0</version>
  </dependency>
  
  <!-- Flink JSON Deserialization Schema -->
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-json</artifactId>
    <version>1.14.0</version>
  </dependency>
</dependencies>

dependencies {
  implementation 'org.apache.flink:flink-connector-kafka_2.12:1.14.0'
  implementation 'org.apache.flink:flink-json:1.14.0'
}

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import java.util.Properties;

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "test-group");

FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("your_topic", new SimpleStringSchema(), properties);
kafkaConsumer.setStartFromLatest(); // 跳过已经消费过的记录

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.addSource(kafkaConsumer).print(); // 打印接收到的数据

// 使用 JSONDeserializer 序列化和反序列化 JSON
import org.apache.flink.formats.json.JsonDeserializer;

// 定义一个 POJO 类来表示 JSON 数据
public class MyJsonData {
  public String field1;
  public int field2;
}

// 创建一个 POJO 类型的 DataStream
DataStream<MyJsonData> jsonDataStream = env.addSource(kafkaConsumer)
  .map(value -> {
    JsonDeserializer<MyJsonData> deserializer = new JsonDeserializer<>(MyJsonData.class);
    return deserializer.deserialize(value);
  });

// 使用 POJO 类型的 DataStream 进行后续处理
jsonDataStream.print();