Flink读写系列之-读Kafka并写入Kafka
程序员文章站
2022-03-08 09:12:39
...
读写Kafka比较简单,官方提供了connector,也提供了例子可以参看,官网例子的GitHub地址:
下面只做简单的说明:
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
}
上一篇: POSIX定时器Timer
下一篇: Apache Flink-下载