在地图方法中发布到Kafka主题

Question

从地图函数（scala）中写入kafka主题？

1. Flink 应用程序中的Kafka主题阅读
2. 在地图功能中处理数据
3. 问题语句 - 在映射功能中，我正在循环浏览列表。对于列表中的每个元素，我想发布到kafka主题。
4. 当我从地图上获取输出并下沉时，它可以工作，但是如果我尝试从地图方法中推出主题，则不会

5. 是否可以在地图方法中发布到主题

   // Main Function
   def main(args: Array[String]) {
   ...
   // some list
   val list_ = ("a", "b", "c", "d")
   // Setup Properties
   val props = new Properties()
   props.setProperty("zookeeper.connect", zookeeper_url + ":" + zookeeper_port)
   props.setProperty("bootstrap.servers", broker_url + ":" + broker_port)
   props.setProperty("auto.offset.reset", "earliest")
   props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
   props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
   ...
   // Connect to Source
   val input_stream = env.addSource(new FlinkKafkaConsumer09[String](topic_in, new SimpleStringSchema(), properties))
   // Process each Record
   val stream = input_stream.map(x=> {        
     // loop through list "list_" -> variable in in Main
     // and publish to topic_out
     // -- THIS IS MY CURRENT ISSUE !!!)
     // -- Does not work (No compile issue)
     // 
     var producer2 = new KafkaProducer[String, String](props)
     var record    = new ProducerRecord(topic_out, "KEY", list(i))
     producer2.send(record)
     producer2.flush()
    // ... Other process and return processed string
   })
   // publish to different topic of proccessed input string (Works)
   stream.addSink(new FlinkKafkaProducer09[String](broker_url + ":" + broker_port, other_topic, new SimpleStringSchema()))
   

Answer 1

不要在地图功能的内部制作kafka生产者，也不要尝试写入地图内的kafka主题。老实说，我不能引用任何话说这是一个坏主意...但这是一个坏主意。

而是。将您的地图功能更改为flatmap（请参阅第一个示例：https：//ci.apache.org/projects/flink/flink/flink/flink-docs-release-1.3/dev/datastream_api.html）。

因此，在您的循环中，您只需要做collector.collect(recordToPublishToKafka)。

，您的水槽将在收集时发布每个人。