link 读取 Kafka多个 Partition问题是什么情况？
我的 Kafka cluster 有三个机器，topic 也分了三个 partition， 我用 Flink 读取 Kafka 消息的时候，感觉只从一个 partition 读取了东西， 一开始我的 bootstrap.servers 只写了一个地址， 现在改成了所有地址，也换了 group.id
Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "10.216.85.201:9092,10.216.77.170:9092,10.216.77.188:9092"); properties.setProperty("group.id", "behavior-logs-aggregator");
FlinkKafkaConsumer010 kafkaConsumer010 = new FlinkKafkaConsumer010 ("behavior-logs_dev", new BehaviorLogDeserializationSchema(), properties); kafkaConsumer010.setStartFromTimestamp(1585670400000L); //2020/04/01
处理完的数据，写到数据库里，看下了感觉少数据， 从 Log 里看到，也是。。，只有 partition=3， 没有 partiton=1，或者 2 的，
2020-04-02 14:54:58,532 INFO org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase - Consumer subtask 0 creating fetcher with offsets {KafkaTopicPartition{topic='behavior-logs_dev', partition=3}=38}.
是哪里有问题吗？*来自志愿者整理的FLINK邮件归档
写回答
-
如果是要排查问题的话可以在消费 kafka 的时候通过 JSONKeyValueDeserializationSchema 来将数据的元数据（topic/parttion/offset）获取，这样可以排查你的数据到底来自哪些分区，这样你就不会再有困惑了。
eg:
env.addSource(new FlinkKafkaConsumer011<>( parameters.get("topic"), new JSONKeyValueDeserializationSchema(true), buildKafkaProps(parameters))) .flatMap(new FlatMapFunction<ObjectNode, ObjectNode>() { @Override public void flatMap(ObjectNode jsonNodes, Collector collector) throws Exception { System.out.println(jsonNodes.get("value")); System.out.println(jsonNodes.get("metadata").get("topic").asText());
System.out.println(jsonNodes.get("metadata").get("offset").asText());
System.out.println(jsonNodes.get("metadata").get("partition").asText()); collector.collect(jsonNodes); } }) .print();*来自志愿者整理的FLINK邮件归档
2021-12-03 19:03:37赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
