java - 如何在 spark streaming 中映射 kafka 主题名称和各自的记录

Question

我正在从 kafka 主题流式传输，如下所示；

JavaPairInputDStream<String, String> directKafkaStream = 
    KafkaUtils.createDirectStream(jssc,
                                  String.class, 
                                  String.class,
                                  StringDecoder.class,
                                  StringDecoder.class,
                                  kafkaParams, 
                                  topicSet);

directKafkaStream.print();   

一个主题的输出如下所示:

(null,"04/15/2015","18:44:14")
(null,"04/15/2015","18:44:15")
(null,"04/15/2015","18:44:16")
(null,"04/15/2015","18:44:17")  

如何映射主题名称和记录。
例如:主题是“callData”，它应该像下面这样

(callData,"04/15/2015","18:44:14")
(callData,"04/15/2015","18:44:15")
(callData,"04/15/2015","18:44:16")
(callData,"04/15/2015","18:44:17")  

Answer 1

How do I map topic name and records?

为了提取分区信息，you'll need to use the overload which accepts a Function receiving MessageAndMetadata<K, V> 并返回您希望转换成的类型。

看起来像这样:

Map<TopicAndPartition, Long> map = new HashMap<>();
map.put(new TopicAndPartition("topicname", 0), 1L);

JavaInputDStream<Map.Entry> stream = KafkaUtils.createDirectStream(
        javaContext,
        String.class,
        String.class,
        StringDecoder.class,
        StringDecoder.class,
        Map.Entry.class, // <--- This is the record return type from the transformation.
        kafkaParams,
        map,
        messageAndMetadata -> 
            new AbstractMap.SimpleEntry<>(messageAndMetadata.topic(),
                                          messageAndMetadata.message()));

注意我用了Map.Entry作为 Java 对 Tuple2 的替代在斯卡拉。您可以提供自己的类(class)，其中有一个 Partition和 Message属性以及用于转换。注意 kafka 输入流的类型现在是 JavaInputDStream<Map.Entry> ，因为这就是转换返回的结果。