scala - 将 Kafka 消费者和生产者集成到一个函数中

Question

我们需要开发一个代码，让消费者运行以监听特定的 kafka 生产者，然后在同一函数中从当前消费的数据生成经过处理的数据到不同的生产者主题。

这是为了将 flinks 代码与 Java 集成，其中 Java 向一个主题生成一条消息，flink 使用它并向另一个主题生成新数据，供 Java 进一步处理。

如果有其他方法可以执行此过程，请告诉我们。

Answer 1

Flink integrates nicely with Kafka ，并且可以在需要时利用 Kafka 事务。这样的应用程序看起来像这样:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
FlinkKafkaConsumer011<> consumer = new FlinkKafkaConsumer011<IN>(topic_in, serializer_in, kafkaProperties);
FlinkKafkaProducer011<> producer = new FlinkKafkaProducer011<OUT>(broker, topic_out, serializer_out)

env.addSource(consumer)
   .map(new SuitableTransformation())
   .addSink(producer)
   .execute()