Flink輸出到Kafka(兩種方式)

方式一：讀取文件輸出到Kafka   

   1.代碼

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011

//溫度傳感器讀取樣例類
case class SensorReading(id: String, timestamp: Long, temperature: Double)

object KafkaSinkTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    import org.apache.flink.api.scala._
    val inputStream = env.readTextFile("sensor.txt")
    val dataStream = inputStream.map(x => {
      val arr = x.split(",")
      SensorReading(arr(0).trim, arr(1).trim.toLong, arr(2).trim.toDouble).toString   //轉成String方便序列化輸出
    })

    //sink
    dataStream.addSink(new FlinkKafkaProducer011[String]("localhost:9092", "sinkTest", new SimpleStringSchema()))
    dataStream.print()

    env.execute(" kafka sink test")

  }
}

2.啟動zookeeper:參考https://www.cnblogs.com/wddqy/p/12156527.html
3.啟動kafka:參考https://www.cnblogs.com/wddqy/p/12156527.html
4.創建kafka消費者觀察結果

方式二：Kafka到Kafka   

   1.代碼

import java.util.Properties
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer011, FlinkKafkaProducer011}

//溫度傳感器讀取樣例類
case class SensorReading(id: String, timestamp: Long, temperature: Double)

object KafkaSinkTest1 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    import org.apache.flink.api.scala._
    //從Kafka到Kafka
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")

    val inputStream = env.addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))
    val dataStream = inputStream.map(x => {
      val arr = x.split(",")
      SensorReading(arr(0).trim, arr(1).trim.toLong, arr(2).trim.toDouble).toString   //轉成String方便序列化輸出
    })

    //sink
    dataStream.addSink(new FlinkKafkaProducer011[String]("localhost:9092", "sinkTest", new SimpleStringSchema()))
    dataStream.print()

    env.execute(" kafka sink test")

  }
}

2.啟動zookeeper:參考https://www.cnblogs.com/wddqy/p/12156527.html
3.啟動kafka:參考https://www.cnblogs.com/wddqy/p/12156527.html
4.創建Kafka生產者和消費者，運行代碼，觀察結果

有幫助的歡迎評論打賞哈，謝謝！