apache spark streaming kafka集成错误

Question

我正在使用火花流和Kafka，我得到了这个错误。

线程“streaming-start”中的异常java.lang.NosuchMethoderror：scala.predef$.arrowassoc（ljava/lang/object；）ljava/lang/object；在org.apache.spark.streaming.kafka010.directkafkainputdstream$$anonfun$start$1.在org.apache.spark.streaming.kafka010.directkafkainputdstream.scala:246)dorg.apache.spark.streaming.kafka010.directkafkainputdstream$$anonfun$start$1.在scala.collection.traversablelike$$anonfun$map$1.在collection.iterator$class.foreach(iterator.scala:727)在Scala.collection.abstractiterator.foreach(iterator.scala:1157)在Scala.collection.iterablelike$class.foreach(iterablelike.scala:72)在Scala.collection.abstractiterable.foreach(iterable.scala:54)在Scala.collection.traversablelike$class.map(traversablelike.scala:244)在.mutable.abstractSet.map(Set.Scala:45)位于org.apache.spark.streaming.kafka010.DirectKafka在org.apache.spark.streaming.dstreamgraph$$anonfun$start$5处Inputdstream.start(drectkafkainputdstream.scala:245)。在org.apache.spark.streaming.dstreamgraph$$anonfun$start$5处应用(dstreamgraph.scala:49)在org.apache.spark.streaming.dstreamgraph$$anonfun$start$5处应用(dstreamgraph.scala:49)在$foreach.leaf(Pariterablelike.scala:975)在Scala.collection.parallel.task$$anonfun$trylefa$1.在Scala.collection.parallel.task$$anonfun$trylefa$1.在Scala.collection.parallel.task$$$anonfun$trylefa$1应用$mcv$sp(tasks.scala:54)在Scala.collection.parallel.task$$anonfun$trylefa$1应用(tasks.scala:53)ection.parallel.AdaptiveWorkStealingTasks$WrappedTask$Class.Compute(tasks.scala:165)at scala.coll在Scala.concurrent.forkjoin.recursiveAction.exec（recursiveAction.java:160)在Scala.concurrent.forkjoin.forkjoinask.doexec（forkjoinask.java:260)在Scala.concurrent.forkjoin.forkjoinpool$workqueue.runtask（forkjoinpool.java:1339)在Scala.concurrent.forkjoinpool.runworker（forkjoinpool.java:1979)在.run（ForkJoinWorkerThread.java:107)17/08/02 16:24:58 INFO StreamingContext：StreamingContext已启动

<dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.10</artifactId>
            <version>2.1.0</version>
        </dependency>
        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>2.8.0</version>
        </dependency>
        <dependency>
            <groupId>com.googlecode.json-simple</groupId>
            <artifactId>json-simple</artifactId>
            <version>1.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>mongo-java-driver</artifactId>
            <version>3.4.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.10</artifactId>
            <version>2.1.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.10</artifactId>
            <version>2.1.1</version>
        </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.10.0.0</version>
    </dependency>

我的代码：

     SparkConf conf = new SparkConf().setAppName("Streaming").setMaster("local");


       JavaStreamingContext streamingContext = new JavaStreamingContext(conf, new Duration(1000));
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", "localhost:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "exastax");
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", false);

        Collection<String> topics = Arrays.asList("loglar");
        JavaInputDStream<ConsumerRecord<String, String>> stream =
                KafkaUtils.createDirectStream(
                        streamingContext,
                        LocationStrategies.PreferConsistent(),
                        ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
                );
stream.foreachRDD(rdd -> {
        OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
        rdd.foreachPartition(consumerRecords -> {
            OffsetRange o = offsetRanges[TaskContext.get().partitionId()];
            System.out.println(
                    o.topic() + " " + o.partition() + " " + o.fromOffset() + " " + o.untilOffset());
        });
    });
        streamingContext.start();
        streamingContext.awaitTermination();
    }
}

我使用KAFKA2.11-0.11.0.0，我尝试搜索这个问题，但我找不到相关的JAR，请帮助我解决这个问题。

Answer 1

您正在混合Scala2.10和Scala2.11代码。在Scala2.10中使用Kafka依赖项，或者在Scala2.11中使用Spark。