如何在 apache 梁的 ParDo 函数中处理 PCollection<KV<String,String>> 中的元素时将元素发布到 kafka 主题？答案

【问题标题】：How to publish elements to a kafka topic while processing the elements in the PCollection<KV<String,String>> in ParDo function in apache beam?如何在 apache 梁的 ParDo 函数中处理 PCollection<KV<String,String>> 中的元素时将元素发布到 kafka 主题？
【发布时间】：2019-06-23 21:03:20
【问题描述】：

我有一个PCollection<KV<String,String>> Pcol，我正在处理ParDo 方法中的每个元素。我想根据某些条件将记录发布到 pardo 本身的 kafka 主题。

我该怎么做？

PCollection<KV<String, String>> Pcol =pipeline.apply("Process Data", 
    ParDo.of(new DoFn<KV<String, String>, KV<String, String>>() {
        @ProcessElement
        public void processElement(ProcessContext c) throws IOException {
            String key = element.getKey();
            if(key==null)
            {//publish to a kafka topic}
        }
    })
);

【问题讨论】：

标签： java apache-kafka apache-beam

【解决方案1】：

在 Beam 中，Sinks 通常由 ParDo 实现。因此，您可以参考如何在 Beam 中实现 Kafka 接收器，以了解如何在 ParDo 中写入 Kafka。更具体地说，这两个类可能会有所帮助：1 和 2。

【讨论】：