添加依賴& API

在pom.xml添加：

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka_2.11</artifactId><version>1.10.1</version> </dependency>

代碼：

package flink_kafkaimport java.util.Properties import org.apache.flink.api.common.serialization.SimpleStringSchema import org.apache.flink.streaming.api.scala._ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerobject MyFlinkKafkaConsumer {def main(args: Array[String]): Unit = {val properties = new Properties()properties.put("bootstrap.servers", "node110:9092,node111:9092,node112:9092")properties.put("group.id", "test")val env = StreamExecutionEnvironment.getExecutionEnvironment//create kafka sourceval kafkaSource = env.addSource(new FlinkKafkaConsumer[String]("demo02",//topicnew SimpleStringSchema(),//seriableproperties//kafka cluster configuration))//SinkkafkaSource.print()//executeenv.execute("read from kafka demo02")} }

運行測試：
① 創建demo02話題，并在demo02寫入數據

② 執行代碼

Flink作為輸出

package flink_kafkaimport java.util.Propertiesimport org.apache.flink.api.common.serialization.SimpleStringSchema import org.apache.flink.streaming.api.scala._ import org.apache.flink.streaming.api.windowing.time.Time import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, FlinkKafkaProducer}object MyFlinkKafkaConsumerAndProducer {def main(args: Array[String]): Unit = {val properties = new Properties()properties.put("bootstrap.servers", "node110:9092,node111:9092,node112:9092")properties.put("group.id", "test")val env = StreamExecutionEnvironment.getExecutionEnvironment//create kafka sourceval kafkaSource = env.addSource(new FlinkKafkaConsumer[String]("demo02",//topicnew SimpleStringSchema(),//seriableproperties//kafka cluster configuration))//transformationval processed = kafkaSource.flatMap(_.split("\\w+")).map((_,1)).keyBy(0).timeWindow(Time.seconds(5)).sum(1).filter(_._2>=3).map(row => row._1+"->"+row._2)//kafka Sinkval kafkaProducer = new FlinkKafkaProducer[String]("demo01",//target topicnew KeyedSerializationSchemaWrapper[String](new SimpleStringSchema()),//seriablization schemaproperties,FlinkKafkaProducer.Semantic.EXACTLY_ONCE)processed.addSink(kafkaProducer)//executeenv.execute("read from kafka demo02 and write to demo01")} }

測試：

總結

以上是生活随笔為你收集整理的学习笔记Flink（七）—— Flink Kafka插件的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。