flink连接kafka用Scala代码实现

以下是Scala代码实现Flink连接Kafka：

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, FlinkKafkaProducer}
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema

object FlinkKafkaExample {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //设置Kafka相关参数
    val kafkaProps = new Properties
    kafkaProps.setProperty("bootstrap.servers", "localhost:9092")
    kafkaProps.setProperty("group.id", "test")

    //创建Kafka source
    val kafkaSource = new FlinkKafkaConsumer[String](
      "topic1",
      new SimpleStringSchema,
      kafkaProps
    )

    //创建Kafka sink
    val kafkaSink = new FlinkKafkaProducer[String](
      "topic2",
      new KeyedSerializationSchema[String] {
        override def serializeKey(element: String): Array[Byte] = null
        override def serializeValue(element: String): Array[Byte] = element.getBytes("UTF-8")
        override def getTargetTopic(element: String): String = null
      },
      kafkaProps
    )

    //读取Kafka数据
    val stream = env.addSource(kafkaSource)

    //处理数据
    val processedStream = stream.map(x => x.toLowerCase())

    //将数据写入Kafka
    processedStream.addSink(kafkaSink)

    env.execute("Flink Kafka Example")

  }

}

代码解释：

- 首先，我们需要获取 `StreamExecutionEnvironment` 对象。

- 然后，我们需要设置 Kafka 相关参数。这些参数包括 `bootstrap.servers`（Kafka 服务器的地址）、`group.id`（消费者组 ID）等。

- 接下来，我们需要创建一个 Kafka 消费者 `FlinkKafkaConsumer` 和一个 Kafka 生产者 `FlinkKafkaProducer`。需要注意的是，在这里我们使用了 `SimpleStringSchema` 作为反序列化器，它可以将 Kafka 中的字符串数据直接转化为 Flink 中的字符串类型。

- 然后，我们创建一个流 `stream`，通过 `addSource` 方法从 Kafka 中读取数据。

- 接下来，我们对读取到的数据进行处理。在这个例子中，我们简单地将所有字母转换为小写。

- 最后，我们把处理后的数据写入 Kafka。使用 `addSink` 方法，将数据流连接到 Kafka 生产者上。

- 最后，我们通过 `env.execute()` 方法启动任务。