我有一个用例，需要将记录从配置单元移动到kafka。我找不到直接将kafka接收器添加到flink数据集的方法。因此，我使用了一种变通方法，在flink数据集上调用了map转换，并且在map函数内部，我给定的记录使用了kafkaProducer.send（）命令。
我面临的问题是我无法在每个工作节点上执行kafkaProducer.flush（），因此用kafka编写的记录数总是比数据集中的记录数少。
是否有一种优雅的方式来处理此问题？我可以在flink中向数据集添加kafka接收器吗？或以某种方式调用kafkaProducer.flush（）作为终结器？
您可以简单地创建一个Sink，它将在后台使用KafkaProducer并将数据写入Kafka。