如何在火花流中映射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

如何映射主题名称和记录？

为了提取分区信息，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 替代 Scala 中的 Tuple2。您可以提供自己的类，该类也具有Partition 和Message 属性，并将 用于转换。请注意，kafka 输入流的类型现在是 JavaInputDStream<Map.Entry>，因为这是转换返回的内容。