使用 Maven 依赖项的 Spark 版本不匹配
Posted
技术标签:
【中文标题】使用 Maven 依赖项的 Spark 版本不匹配【英文标题】：Spark version mismatch using maven dependencies 【发布时间】：2017-11-28 11:56:54 【问题描述】：我希望 ro 使用 apache Spark 运行简单的 worcount ecample。在 $SPARK_HOME/jars 中使用本地 jar 文件可以正常运行，但使用 maven 依赖会出错：
java.lang.NoSuchMethodError: org.apache.hadoop.fs.FileSystem$Statistics.getThreadStatistics()Lorg/apache/hadoop/fs/FileSystem$Statistics$StatisticsData;
at org.apache.spark.deploy.SparkHadoopUtil$$anonfun$1$$anonfun$apply$mcJ$sp$1.apply(SparkHadoopUtil.scala:149)
at org.apache.spark.deploy.SparkHadoopUtil$$anonfun$1$$anonfun$apply$mcJ$sp$1.apply(SparkHadoopUtil.scala:149)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.Iterator$class.foreach(Iterator.scala:893)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at org.apache.spark.deploy.SparkHadoopUtil$$anonfun$1.apply$mcJ$sp(SparkHadoopUtil.scala:149)
at org.apache.spark.deploy.SparkHadoopUtil.getFSBytesReadOnThreadCallback(SparkHadoopUtil.scala:150)
at org.apache.spark.rdd.HadoopRDD$$anon$1.<init>(HadoopRDD.scala:224)
at org.apache.spark.rdd.HadoopRDD.compute(HadoopRDD.scala:203)
at org.apache.spark.rdd.HadoopRDD.compute(HadoopRDD.scala:94)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
at org.apache.spark.scheduler.Task.run(Task.scala:108)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748)
代码如下：
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;
import java.util.Arrays;
public class SparkTest
public static void main(String[] args)
SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("SparkTest");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaRDD<String> rdd = sc.textFile("file:///usr/local/spark/LICENSE");
JavaPairRDD<String, Integer> counts = rdd
.flatMap(s -> Arrays.asList(s.split(" ")).iterator())
.mapToPair(word -> new Tuple2<>(word, 1))
.reduceByKey((a, b) -> a + b);
counts.coalesce(1).saveAsTextFile("file:///home/XXX/Desktop/Processing/spark");
这里是POM.xml 文件：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>Processing</groupId>
<artifactId>Streaming</artifactId>
<version>1.0-SNAPSHOT</version>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.3.2</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>1.3.2</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>1.3.2</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.10.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka-0.10_2.11</artifactId>
<version>1.3.2</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.7.3</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-filesystem_2.11</artifactId>
<version>1.3.2</version>
</dependency>
</dependencies>
</project>
它还包括一些其他的 apache 软件，如 Hadoop 和 Flink。
已安装 Spark 版本：2.2.0
下载链接：https://www.apache.org/dyn/closer.lua/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz
Hadoop 安装版本 = 2.7.3
这里有些不匹配！
【问题讨论】：
也从您的 pom 中发布与 hadoop 相关的依赖项。 @philantrovert 我的代码取决于 Hadoop 和官方 Spark 网站In addition, if you wish to access an HDFS cluster, you need to add a dependency on hadoop-client for your version of HDFS.
@philantrovert Hadoop 依赖项已添加但未使用。
请提供您的 pom.xml 中的所有依赖项，它似乎来自您的一个依赖项
@PierreB。谢谢，已更新。
【参考方案1】：
使用您的依赖项并展示 Java 如何使用 org.apache.hadoop.fs.FileSystem.class.getResource("FileSystem.class") 加载您的类，您的 jar 似乎是从 org.apache.flink:flink-shaded-hadoop2:jar:1.3.2 加载的。当使用mvn dependency:tree 显示依赖关系树时，我们看到它是来自flink-java: 和flink-streaming-java_2.11 的传递依赖关系
[INFO] +- org.apache.flink:flink-java:jar:1.3.2:compile
[INFO] | +- ...
[INFO] | +- org.apache.flink:flink-shaded-hadoop2:jar:1.3.2:compile
[INFO] +- org.apache.flink:flink-streaming-java_2.11:jar:1.3.2:compile
[INFO] | +- org.apache.flink:flink-runtime_2.11:jar:1.3.2:compile
[INFO] | | +- org.apache.flink:flink-shaded-hadoop2:jar:1.3.2:compile
此 jar 包含整个 org.apache.hadoop.fs 包，覆盖正确的定义并导致您的问题。您可以尝试删除 flink-java 依赖项或排除 flink-shaded-hadoop2 ，但这可能会导致您的代码出现问题，因为可能缺少其他必需的 Flink 类。 例如：
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.3.2</version>
<exclusions>
<exclusion>
<groupId>org.apache.flink</groupId>
<artifactId>flink-shaded-hadoop2</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>1.3.2</version>
<exclusions>
<exclusion>
<groupId>org.apache.flink</groupId>
<artifactId>flink-shaded-hadoop2</artifactId>
</exclusion>
</exclusions>
</dependency>
否则你将不得不根据你的项目需求寻找另一个解决方案：玩转类加载以确保你的类被正确加载，更新你的依赖版本以使 Hadoop 类与 Flink 匹配，等等。
【讨论】：
【参考方案2】：最后为 Spark 创建另一个专用的 maven 项目，只需 spark-core maven 依赖即可工作。
谁能说出原因？
【讨论】：
使用你的依赖并展示Java如何用org.apache.hadoop.fs.FileSystem.class.getResource("FileSystem.class")加载你的类，它出现org.apache.flink:flink-java:jar:1.3.2定义一个传递依赖org.apache.flink:flink-shaded-hadoop2:jar:1.3.2，它本身包含整个org.apache.hadoop.fspackage...你的应用程序加载来自这个伪造的 jar 的类，而不是正确的，因此你的错误。使用 Spark Core 作为主要依赖似乎是一种适当的解决方法，但如果您可以完全排除 flink-shaded-hadoop2 而不会出现更好的问题
@PierreB。伟大的！发表答案，我会得到正确答案
很高兴有帮助，我发布了答案【参考方案3】：
从 Flink 1.4（发布待定）开始，Flink 可以在没有任何 hadoop 依赖项的情况下运行，如果您需要 hadoop，类路径中包含 hadoop 就足够了。这应该会让您的生活更轻松。
【讨论】：
以上是关于使用 Maven 依赖项的 Spark 版本不匹配的主要内容，如果未能解决你的问题，请参考以下文章
gradle 到 maven 的转换，在依赖项的依赖项的版本名称中解析通配符“+”？