Flink1.7.2数据集不支持kafka接收器吗?完成批处理操作后,我需要将消息发布到kafka,意思是源是我的postgres,而sink是我的kafka。有可能吗?
8ljdwjyq1#
开箱即用:还没有,您必须从头开始使用数据流,或者创建您自己的自定义输出格式。不过,有计划要统一这个国家 DataSet 以及 DataStream ApacheFlink项目中的API在flink 2.0中的长期应用:https://flink.apache.org/roadmap.html
DataSet
DataStream
w41d8nur2#
您可以创建自己的输出格式,并使用Kafka生产者生产Kafka。检查下面的代码。
... data.output(new KafkaOPFormat()); env.execute();
import java.io.IOException; import java.util.Properties; import org.apache.flink.api.common.io.RichOutputFormat; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; public class KafkaOPFormat extends RichOutputFormat<Tuple2<String, String>> { private final Properties properties = new Properties(); private KafkaProducer<String, String> producer; @Override public void configure(Configuration configuration) { properties.put("bootstrap.servers", "localhost:9092"); properties.put("acks", "all"); properties.put("retries", 0); properties.put("batch.size", 16384); properties.put("linger.ms", 1); properties.put("buffer.memory", 33554432); properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); } @Override public void open(int i, int i1) throws IOException { producer = new KafkaProducer<String, String>(properties); } @Override public void writeRecord(Tuple2<String, String> record) throws IOException { producer.send(new ProducerRecord<>(record.f0, record.f1)); } @Override public void close() throws IOException { producer.close(); } }
ps:我不记得所有的配置,请检查您的配置并相应地更改。
2条答案
按热度按时间8ljdwjyq1#
开箱即用:还没有,您必须从头开始使用数据流,或者创建您自己的自定义输出格式。
不过,有计划要统一这个国家
DataSet
以及DataStream
ApacheFlink项目中的API在flink 2.0中的长期应用:https://flink.apache.org/roadmap.htmlw41d8nur2#
您可以创建自己的输出格式,并使用Kafka生产者生产Kafka。检查下面的代码。
ps:我不记得所有的配置,请检查您的配置并相应地更改。