Spark 和 Kafka 问题 - 线程“main”中的异常 java.lang.NoClassDefFoundError: org.apache.spark.streaming.kafka010.LocationStrategies

Question

我首先要提到的是，我已经尝试了类似主题中的所有建议，但对我没有任何帮助，所以请这不是重复的问题。

我遇到的问题如下 -

我正在尝试使用 spark 流和 kafka 在 spark 上运行示例 java 应用程序。我已经添加了所有必需的依赖项：

  <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.11.12</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>
    </dependencies>

在我想运行我的应用程序的服务器上部署 Jar 之后（我已经使用 spark、kafka 设置了环境并创建了相关主题）我正在尝试 spark-submit 它并收到以下错误：

Exception in thread "main" java.lang.NoClassDefFoundError: org.apache.spark.streaming.kafka010.LocationStrategies
        at JavaWordCount.main(JavaWordCount.java:47)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:90)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:55)
        at java.lang.reflect.Method.invoke(Method.java:508)
        at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
        at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:879)
        at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:197)
        at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:227)
        at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:136)
        at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.ClassNotFoundException: org.apache.spark.streaming.kafka010.LocationStrategies
        at java.net.URLClassLoader.findClass(URLClassLoader.java:609)
        at java.lang.ClassLoader.loadClassHelper(ClassLoader.java:924)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:869)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:852)

似乎工作人员无法将依赖项识别为环境的一部分。我在网上做了一些研究，许多人建议使用 maven-shade-plugin 创建一个程序集 JAR。所以我也尝试用这种方式对jar进行maven-package，但仍然没有成功。

作为参考，这里是应用失败的地方：

// Configure Spark to connect to Kafka running on local machine
Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
kafkaParams.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
                "org.apache.kafka.common.serialization.StringDeserializer");
kafkaParams.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
                "org.apache.kafka.common.serialization.StringDeserializer");
kafkaParams.put(ConsumerConfig.GROUP_ID_CONFIG,"group1");
kafkaParams.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
kafkaParams.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);

//Configure Spark to listen messages in topic test
Collection<String> topics = Arrays.asList("wordCount");

SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("SparkKafkaWordCount");

//Read messages in batch of 30 seconds
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(30));

// Start reading messages from Kafka and get DStream
final JavaInputDStream<ConsumerRecord<String, String>> stream =
        KafkaUtils.createDirectStream(jssc, LocationStrategies.PreferConsistent(), 
                                      ConsumerStrategies.<String,String>Subscribe(topics,kafkaParams));

在上面的最后一行中，LocationStrategies 类无法识别，即使我已在 pom.xml 中添加了正确的依赖项

有什么办法可以解决这个问题吗？

Answer 1

将此添加到您的 pom.xml 中

        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-assembly-plugin</artifactId>

            <configuration>
                <archive>
                    <manifest>
                        <mainClass>your main class </mainClass>
                    </manifest>
                </archive>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
        </plugin>

然后用

构建jar

mvn clean compile assembly:single

你应该在目标目录下得到两个jar，一个没有依赖，一个有依赖（your-jar-1.0-jar-with-dependencies.jar）

Answer 2

当我在 --jar 下的 spark-spark-submit 命令中包含以下 jar 时，我没有收到此错误：

• sql-kafka-0-10_2.11-2.2.1.jar
• kafka-clients-0.10.1.0.jar
• spark-streaming-kafka-0-10_2.11-2.2.1.jar
• spark-streaming-kafka-0-10-assembly_2.11-2.2.1.jar
• spark-streaming-kafka_2.11-1.6.3.jar

Answer 3

即使我遇到了同样的问题，谷歌搜索也没有得到太多帮助，但是我在阅读了许多线程后了解到，pom.xml 文件中提到的依赖项的范围为“已提供”，这意味着我们需要在执行期间指定依赖的jar文件。此外，Apache Spark 包中的所有示例都被编译为单个 jar 文件，我们需要指定类路径来执行所需的模块。下载你在 pom.xml 中提到的必要的 jar 文件并像这样执行

spark-submit --jars kafka-clients-1.1.0.jar,spark-streaming_2.11-2.3.0.jar,spark-streaming-kafka-0-10_2.11-2.3.0.jar --class org.apache.spark.examples.streaming.JavaDirectKafkaWordCount target/original-spark-examples_2.11-2.4.0-SNAPSHOT.jar <brokerip:port> <topic>

但在此之前，您需要在 java 文件中重新编写使用者属性，否则您将收到错误提示配置丢失

kafkaParams.put("bootstrap.servers", brokers);
kafkaParams.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
kafkaParams.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
kafkaParams.put("group.id", "<group_id>");