kafka引发流式java api问题

Question

我是Kafka和spark的初学者。我想通过spark streaming对我从Kafka收到的特定主题的数据进行实时处理。我无法使用createStream函数返回的JavaPairReceiverInputStream。

SparkConf conf = new SparkConf().setMaster("local[2]").setAppName(
                "testwordCount");
        JavaStreamingContext jssc = new JavaStreamingContext(conf,
                Durations.seconds(1));

        Map<String, Integer> topics_map = new HashMap<String, Integer>();

        topics_map.put("Customtopic", 10);

JavaPairReceiverInputDStream<String, String> kafkaStream = KafkaUtils
                .createStream(jssc, "localhost:2181", "kafkasparkconsumer",
                        topics_map);

下面的代码给出了一个错误：

 JavaPairDStream<String, Integer> wordCounts = kafkaStream.map(
                 new PairFunction<String, String, Integer>() {
                 @Override public Tuple2<String, Integer> call(String s) {
                 return new Tuple2<String, Integer>(s, 1);
                 }
                 }).reduceByKey(new Function2<Integer, Integer, Integer>() {
                 @Override
                 public Integer call(Integer i1, Integer i2) {
                 return i1 + i2;
                 }
                 });
                 wordCounts.print();

方法图（Function， R

我使用的spark版本是1.2.0。我找不到处理Kafka消息的java api示例。有人能告诉我我需要改变什么吗？

Answer 1

kafkaStream返回一个元组。检查这个

JavaPairReceiverInputDStream<String, String> kafkaStream = KafkaUtils
                .createStream(jssc, "localhost:2181", "kafkasparkconsumer",
                        topics_map);
JavaDStream<String> lines = kafkaStream
                .map(new Function<Tuple2<String, String>, String>() {
                    /**
                     * 
                     */
                    private static final long serialVersionUID = 1L;

                    @Override
                    public String call(Tuple2<String, String> tuple2) {
                        return tuple2._2();
                    }
                });

Answer 2

您调用了错误的方法。在java中，如果您想获得一对，应该调用MapToPair。请尝试以下代码：

JavaPairDStream<String, Integer> pairs = kafkaStream
            .mapToPair(new PairFunction<Tuple2<String, String>, String, Integer>() {
                @Override public Tuple2<String, Integer> call(Tuple2<String, String> word) throws Exception {
                    return new Tuple2<String, Integer>(word._2(), 1);
                }
            }).reduceByKey(new Function2<Integer, Integer, Integer>() {
                @Override public Integer call(Integer integer, Integer integer2) throws Exception {
                    return integer + integer2;
                }
            });

    pairs.print();

    jssc.start();
    jssc.awaitTermination();