Flink常用API之Kafka的Source

2023年5月25日下午11:52 • 大数据 • 阅读 97

Flink常用API之Kafka的Source

原创

wx62be9d88ce2942022-07-01 17:44:41博主文章分类：大数据 ©著作权

文章标签 kafka flink scala apache 文章分类 Hadoop 大数据

©著作权归作者所有：来自51CTO博客作者wx62be9d88ce294的原创作品，请联系作者获取转载授权，否则将追究法律责任

基于 Kafka 的 Source
首先需要配置 Kafka 连接器的依赖，另外更多的连接器可以查看官网： https://ci.apache.org/projects/flink/flink-docs-release-1.9/zh/dev/connectors/

<dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-connector-kafka_2.11artifactId>
            <version>1.9.1version>
        dependency>

第一种：读取 Kafka 中的普通数据（String）

package sourceimport org.apache.flink.api.common.serialization.SimpleStringSchemaimport org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentimport org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerimport org.apache.kafka.common.serialization.StringDeserializerimport java.util.Properties    object KafkaSource {  def main(args: Array[String]): Unit = {    val environment = StreamExecutionEnvironment.getExecutionEnvironment    environment.setParallelism(1)    import org.apache.flink.streaming.api.scala._        val properties = new Properties()    properties.setProperty("bootstrap.servers","node1:9092,node2:9092,node3:9092")    properties.setProperty("group.id","flink01")    properties.setProperty("key.deserializer",classOf[StringDeserializer].getName)    properties.setProperty("value.deserializer",classOf[StringDeserializer].getName)    properties.setProperty("auto.offset.reset","latest")    environment.addSource(new FlinkKafkaConsumer[String]("topic_01",new SimpleStringSchema(),properties))      .print()    environment.execute()  }}

第二种：读取 Kafka 中的KeyValue数据
定义消费者

package sourceimport org.apache.flink.api.common.typeinfo.TypeInformationimport org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTuple2TypeInformation, createTypeInformation}import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, KafkaDeserializationSchema}import org.apache.kafka.clients.consumer.ConsumerRecordimport org.apache.kafka.common.serialization.StringDeserializerimport java.util.Properties    import org.apache.flink.streaming.api.scala._object KafkaSourceKeyValue {      def main(args: Array[String]): Unit = {            val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment            streamEnv.setParallelism(1)             val props = new Properties()      props.setProperty("bootstrap.servers","hadoop101:9092,hadoop102:9092,hadoop103:9092")      props.setProperty("group.id","flink01")      props.setProperty("key.deserializer",classOf[StringDeserializer].getName)      props.setProperty("value.deserializer",classOf[StringDeserializer].getName)      props.setProperty("auto.offset.reset","latest")            val stream: DataStream[(String, String)] = streamEnv.addSource(new FlinkKafkaConsumer[(String,String)]("t_bjsxt",new MyKafkaReader,props))      stream.print()      streamEnv.execute()    }        class MyKafkaReader extends  KafkaDeserializationSchema[(String,String)]{            override def isEndOfStream(nextElement: (String, String)): Boolean = {        false      }            override def deserialize(record: ConsumerRecord[Array[Byte], Array[Byte]]): (String, String) = {        if(record!=null){          var key="null"          var value="null"          if(record.key()!=null){            key =new String(record.key(),"UTF-8")          }          if(record.value()!=null){             value =new String(record.value(),"UTF-8")          }          (key,value)        }else{          ("null","nulll")        }      }            override def getProducedType: TypeInformation[(String, String)] ={        createTuple2TypeInformation(createTypeInformation[String],createTypeInformation[String])      }    }}

定义生产者

package sourceimport org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}import java.util.{Properties, Random}    object MykafkaProducer {  def main(args: Array[String]): Unit = {        val props = new Properties()    props.setProperty("bootstrap.servers","node1:9092,node2:9092,node3:9092")    props.setProperty("key.serializer",classOf[StringSerializer].getName)    props.setProperty("value.serializer",classOf[StringSerializer].getName)    var producer =new KafkaProducer[String,String](props)    var r =new Random()    while(true){       val data = new ProducerRecord[String,String]("topic_01","key"+r.nextInt(10),"value"+r.nextInt(100))      producer.send(data)      Thread.sleep(1000)    }    producer.close()  }}