使用ApacheFlink仅生成Kafka主题的消息

Question

从示例中，我看到了下面的代码片段，它运行良好。但问题是：我并不总是需要处理输入流并将其生成到接收器。

如果我有一个应用程序，根据某些事件，我必须只发布到kafka主题，以便下游应用程序可以做出某些决定。这意味着，我实际上没有输入流，但我只知道当我的应用程序中发生某些事情时，我需要向kafka的特定主题发布消息。也就是说，我只需要一个接收器。

我查看了示例，但没有找到符合我要求的任何内容。有没有一种方法可以只配置一个KafkaSink，该KafkaSink公开一个方法()来发布消息到一个主题。

提前表示感谢！！

String inputTopic = "flink_input";
    String outputTopic = "flink_output";
    String consumerGroup = "baeldung";
    String address = "localhost:9092";
    StreamExecutionEnvironment environment = StreamExecutionEnvironment
      .getExecutionEnvironment();
    FlinkKafkaConsumer011<String> flinkKafkaConsumer = createStringConsumerForTopic(
      inputTopic, address, consumerGroup);
    DataStream<String> stringInputStream = environment
      .addSource(flinkKafkaConsumer);

    FlinkKafkaProducer011<String> flinkKafkaProducer = createStringProducer(
      outputTopic, address);

    stringInputStream
      .map(new WordsCapitalizer())
      .addSink(flinkKafkaProducer);

Answer 1

您必须有一个来源。您可能希望实现自定义源，或者可以使用类似 NumberSequenceSource 的东西，后跟一个运算符（如 process 函数），该函数发出您知道要写入接收器的任何内容，然后是接收器。

例如，该流程函数可以将传入事件转换为您想要写入Kafka的任何内容，或者它可以忽略其输入并使用计时器生成要发送给Kafka（Kafka）的事件。

或者，根据您的需求，您可能会发现异步i/o是比流程函数更好的构建块。