在 Flink Kafka Producer 中发送密钥
Send key in Flink Kafka Producer
我是 Flink Stream 处理的新手,需要一些 Flink Kafka producer 方面的帮助,因为经过一段时间的搜索后找不到太多与之相关的内容。我目前正在从 Kafka 主题读取流,然后在执行一些计算后我想将它写到 Kafka 中的一个新的单独主题。但我面临的问题是我无法将密钥发送到 Kafka 主题。我正在使用 Flink Kafka 连接器,它为我提供了 FlinkKafkaConsumer 和 FlinkKafkaProducer。要更详细地看下面是我的代码,我可以在我的代码中更改什么它可以工作,目前在 Kafka 上我正在生成我的消息在 Key 中使用 null,因为值是我需要的:
Properties consumerProperties = new Properties();
consumerProperties.setProperty("bootstrap.servers", serverURL);
consumerProperties.setProperty("group.id", groupID);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(consumerTopicName,
new SimpleStringSchema(), consumerProperties);
kafkaConsumer.setStartFromEarliest();
DataStream<String> kafkaConsumerStream = env.addSource(kafkaConsumer);
final int[] tVoteCount = {0};
DataStream<String> kafkaProducerStream = kafkaConsumerStream.map(new MapFunction<String, String>() {
@Override
public String map(String value) throws InterruptedException, IOException {
JsonNode jsonNode = jsonParser.readValue(value, JsonNode.class);
Tcount = Tcount + jsonNode.get(key1).asInt();
int nameCandidate = jsonNode.get(key2).asInt();
System.out.println(Tcount);
String tCountT = Integer.toString(Tcount);
//tVoteCount = tVoteCount + voteCount;
//waitForEventTime(timeStamp);
return tCountT;
}
});
kafkaConsumerStream.print();
System.out.println("sdjknvksjdnv"+Tcount);
Properties producerProperties = new Properties();
producerProperties.setProperty("bootstrap.servers", serverURL);
FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(producerTopicName,
new SimpleStringSchema(), producerProperties);
kafkaProducerStream.addSink(kafkaProducer);
env.execute();
谢谢。
在此 blog 中,您将找到有关如何将密钥和主题写入主题的示例:
您需要将您创建的 new FlinkKafkaProducer
替换为如下内容:
FlinkKafkaProducer<KafkaRecord> kafkaProducer =
new FlinkKafkaProducer<KafkaRecord>(
producerTopicName,
((record, timestamp) -> new ProducerRecord<byte[], byte[]>(producerTopicName, record.key.getBytes(), record.value.getBytes())),
producerProperties
);
如果您提供自己的 KafkaSerializationSchema 而不是使用 SimpleStringSchema
,那么您将可以完全控制所写的内容。 @mike 在他的回答中提供了一个如何做到这一点的例子。
我是 Flink Stream 处理的新手,需要一些 Flink Kafka producer 方面的帮助,因为经过一段时间的搜索后找不到太多与之相关的内容。我目前正在从 Kafka 主题读取流,然后在执行一些计算后我想将它写到 Kafka 中的一个新的单独主题。但我面临的问题是我无法将密钥发送到 Kafka 主题。我正在使用 Flink Kafka 连接器,它为我提供了 FlinkKafkaConsumer 和 FlinkKafkaProducer。要更详细地看下面是我的代码,我可以在我的代码中更改什么它可以工作,目前在 Kafka 上我正在生成我的消息在 Key 中使用 null,因为值是我需要的:
Properties consumerProperties = new Properties();
consumerProperties.setProperty("bootstrap.servers", serverURL);
consumerProperties.setProperty("group.id", groupID);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(consumerTopicName,
new SimpleStringSchema(), consumerProperties);
kafkaConsumer.setStartFromEarliest();
DataStream<String> kafkaConsumerStream = env.addSource(kafkaConsumer);
final int[] tVoteCount = {0};
DataStream<String> kafkaProducerStream = kafkaConsumerStream.map(new MapFunction<String, String>() {
@Override
public String map(String value) throws InterruptedException, IOException {
JsonNode jsonNode = jsonParser.readValue(value, JsonNode.class);
Tcount = Tcount + jsonNode.get(key1).asInt();
int nameCandidate = jsonNode.get(key2).asInt();
System.out.println(Tcount);
String tCountT = Integer.toString(Tcount);
//tVoteCount = tVoteCount + voteCount;
//waitForEventTime(timeStamp);
return tCountT;
}
});
kafkaConsumerStream.print();
System.out.println("sdjknvksjdnv"+Tcount);
Properties producerProperties = new Properties();
producerProperties.setProperty("bootstrap.servers", serverURL);
FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(producerTopicName,
new SimpleStringSchema(), producerProperties);
kafkaProducerStream.addSink(kafkaProducer);
env.execute();
谢谢。
在此 blog 中,您将找到有关如何将密钥和主题写入主题的示例:
您需要将您创建的 new FlinkKafkaProducer
替换为如下内容:
FlinkKafkaProducer<KafkaRecord> kafkaProducer =
new FlinkKafkaProducer<KafkaRecord>(
producerTopicName,
((record, timestamp) -> new ProducerRecord<byte[], byte[]>(producerTopicName, record.key.getBytes(), record.value.getBytes())),
producerProperties
);
如果您提供自己的 KafkaSerializationSchema 而不是使用 SimpleStringSchema
,那么您将可以完全控制所写的内容。 @mike 在他的回答中提供了一个如何做到这一点的例子。