zu kafka Thema innerhalb einer Kartenmethode veröffentlichen

Question

Schreiben zu kafka Thema von innerhalb einer Kartenfunktion (SCALA)?

1. von einem kafka Thema in einer Anwendung FLINK Lesen
2. die Daten innerhalb einer Map-Funktion bearbeitet
3. Problemstellung - Innerhalb der Karte Funktion, die ich durch eine Liste bin Looping. Für jedes Element in der Liste möchte ich ein Kafka-Thema veröffentlichen.
4. wenn ich die Ausgabe von der Karte und Spüle es funktioniert, aber wenn ich eine Push-to-Thema aus den Karten Methode versuchen tut es nicht

5. Ist es möglich, aus einer Karte Methode Thema zu veröffentlichen

   // 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

Sie keinen kafka Produzenten innerhalb einer Kartenfunktion machen und versuchen Sie nicht innerhalb von einer Karte auf einen kafka Thema zu schreiben. Ehrlich gesagt, ich kann nichts sagen, es ist eine schlechte Idee ... aber es ist eine schlechte Idee.

Stattdessen. Ändern Sie Ihre Kartenfunktion in eine FlatMap (siehe das erste Beispiel hier: https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/datastream_api.html).

Also in Ihrer Schleife, anstatt einen Kafka-Produzent jede Schleife zu machen, tun Sie einfach collector.collect(recordToPublishToKafka).

Und Ihre Spüle wird jedes herausgeben, wie sie gesammelt werden.