如何使用Apache Flink 1.12和DataStream API批处理模式将Kafka添加为有界源

Question

我想在Apache Flink 1.12中使用Kafka源作为有界数据源，我尝试过使用Flink Kafka消费者连接器，但它给了我以下原因

原因：java.lang.IllegalStateException：检测到一个未绑定的源，execution.runtime模式设置为BATCH。不允许此组合，请在org.apache.flink.util.Preconditions.check状态下将execution.runtime模式设置为STREAMING或AUTOMATIC（Preconditions.java:198）~[flink-core-1.12.0.jar： 1.12.0]

根据flink的最新文档，我们可以使用Kafka作为有界源，但没有提供如何实现这一点的示例，也没有提到这是继续使用此方法的最佳方式。

有人能帮我获得一些示例工作代码来实现这个用例吗

Answer 1

这里有一个例子：

KafkaSource<String> source = KafkaSource
        .<String>builder()
        .setBootstrapServers(...)
        .setGroupId(...)
        .setTopics(...)
        .setDeserializer(...)
        .setStartingOffsets(OffsetsInitializer.earliest())
        .setBounded(OffsetsInitializer.latest())
        .build();

env.fromSource(source, WatermarkStrategy.forMonotonousTimestamps(), "Kafka Source"));

有关更多信息，请参阅javadocs。