欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Flink读写系列之-读Kafka并写入Kafka

程序员文章站 2022-03-08 09:12:39
...

读写Kafka比较简单,官方提供了connector,也提供了例子可以参看,官网例子的GitHub地址:

https://github.com/apache/flink/tree/master/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/kafka

下面只做简单的说明:

    val myConsumer = new FlinkKafkaConsumer010[String](KAFKA_CONSUMER_TOPIC, new SimpleStringSchema(), KAFKA_PROP)//参数分别是:指定取数据的topic,序列化器,kafka参数
    env.addSource[String](myConsumer)//读入kafka数据产生dataStream,可针对dataStream进行各种逻辑处理,比如map,filter等等。

//处理结果写入kafka
 dataStream.addSink(new FlinkKafkaProducer010[String](KAFKA_PRODUCER_TOPIC,new SimpleStringSchema(),KAFKA_PROP))//参数分别是:写入topic,序列化器,kafka配置惨

//kafka配置参数
  lazy val KAFKA_PROP: Properties = new Properties() {
    setProperty("bootstrap.servers", KAFKA_BROKER)//broker地址
    setProperty("zookeeper.connect", KAFKA_ZOOKEEPER_HOST)//zookeeper配置
    setProperty("group.id", KAFKA_GROUP_ID)//组id
  }

 

相关标签: flink