scalaflink1.14消费kafkajson數據

以下是使用Scala和Flink 1.14消费Kafka JSON数据的示例代码：

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

object KafkaJSONConsumer {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
  
    val properties = new java.util.Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("group.id", "test-group")
    properties.setProperty("auto.offset.reset", "earliest")
  
    val kafkaConsumer = new FlinkKafkaConsumer[String]("test-topic", new SimpleStringSchema(), properties)
  
    val stream = env.addSource(kafkaConsumer)
  
    val jsonStream = stream.map(json => {
      import org.json4s._
      import org.json4s.jackson.JsonMethods._
  
      implicit val formats = DefaultFormats
      parse(json).extract[MyClass]
    })
  
    jsonStream.print()
  
    env.execute("Kafka JSON Consumer")
  }
  
  case class MyClass(id: Int, name: String, age: Int)
}

在代码中，我们首先创建了一个`StreamExecutionEnvironment`对象。然后，我们定义了一个`Properties`对象，用于配置Kafka连接和消费者组ID。接下来，我们创建了一个`FlinkKafkaConsumer`对象，并将其添加到流中。然后，我们使用`map`操作符将从Kafka读取的JSON字符串转换为`MyClass`对象。最后，我们使用`print`操作符将流打印到控制台。

请注意，我们在代码中使用了`json4s`和`json4s-jackson`库来解析JSON字符串。您可以使用其他JSON解析库，例如`Jackson`或`Gson`。

希望这可以帮助您开始使用Scala和Flink消费Kafka JSON数据。