客户纱线作业上的 Scala 错误

Question

我知道对此存在一些问题，但没有足够的信息来解决我的问题。

我尝试从我的 Eclipse 项目中以 yarn-client 模式运行作业。我有一个带有 2 个节点的 hadoop 集群（其中一个节点当前处于关闭状态）。我尝试在集群模式下运行它（使用 spark-submit）并且它可以工作。我尝试从 Eclipse 项目本地运行它：

我正在尝试制作这样的 Spark 上下文：

        SparkConf conf = new SparkConf().setAppName("AnomalyDetection-BuildModel").setMaster("local[*]");

它的工作原理。

但是当我尝试使用“yarn-client”运行它时：

 SparkConf conf = new SparkConf().setAppName("AnomalyDetection-BuildModel").setMaster("yarn-client").set("driver-memory", "556m").set("executor-memory", "556m").set("executor-cores", "1").set("queue", "default");

我收到一个错误：

cannot assign instance of scala.collection.immutable.List$SerializationProxy to field org.apache.spark.rdd.RDD.org$apache$spark$rdd$RDD$$dependencies_ of type scala.collection.Seq in instance of org.apache.spark.rdd.MapPartitionsRDD

另一个问题是我不知道在这种情况下依赖和兼容性是如何工作的，以及为什么使用 local[*] 我没有收到任何错误。

这是我的 pom.xml 文件：

<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>buildModelTest</groupId>
    <artifactId>buildModelTest</artifactId>
    <version>1</version>
    <properties>
        <encoding>UTF-8</encoding>
        <scala.version>2.11.8</scala.version>
        <spark.version>2.1.0</spark.version>
        <hadoop.version>2.7.0</hadoop.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>3.8.1</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.10</artifactId>
            <version>2.1.0</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_2.10</artifactId>
            <version>2.1.0</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-reflect</artifactId>
            <version>2.11.8</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-yarn_2.10</artifactId>
            <version>2.1.0</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.scalatest</groupId>
            <artifactId>scalatest_2.11</artifactId>
            <version>3.0.0</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>buildModelTest.Main</mainClass>
                      </manifest>
                    </archive>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.3</version>
            </plugin>
        </plugins>
    </build>
</project>

在 Eclipse 项目中，我添加了 hadoop 的配置文件，并在构建配置时添加了 SCALA_HOME、SPARK_HOME、HADOOP_CONF_DIR 的环境变量。关于 SPARK，我有 spark-2.1.0-bin-hadoop2.7 和 SCALA 2.11.8。在我的 Java 项目中，我添加了 Spark/bin 中的所有 jar。

那么你们知道为什么这不适用于“client-yarn”，是否存在依赖性问题？如果是的话，在依赖方面，普通客户端和纱线客户端有什么不同？ Maven 为我下载了一些我从 Spark/bin 添加的 jar，所以我想其中一些是多余的。

编辑

sparkContext 已正确初始化（我猜）。当我调用 .rrd() 方法时抛出错误：

       JavaRDD<Vector> parsedTrainingData = data.map(new Function<String, Vector>() {

                    private static final long serialVersionUID = 1L;

                    public Vector call(String s) {
                        String[] sarray = s.split(" ");
                        double[] values = new double[sarray.length];
                        for (int i = 0; i < sarray.length; i++) {
                            values[i] = Double.parseDouble(sarray[i]);
                        }
                        return Vectors.dense(values);
                    }
                });
                parsedTrainingData.cache();

                // Cluster the data into two classes using KMeans
                KMeansModel clusters = KMeans.train(parsedTrainingData.rdd(), numClusters, numIterations);

Answer 1

从您的代码来看，您似乎正试图在 YARN 集群上运行您的 Spark 应用程序。

SparkConf conf = new SparkConf().setAppName("AnomalyDetection-BuildModel").setMaster("yarn-client").set("driver-memory", "556m").set("executor-memory", "556m").set("executor-cores", "1").set("queue", "default");

这里，setMaster("yarn-client") 是 master 的错误参数。

当您将 master 设置为 local[*] 时，您的 spark 应用程序在您机器的单个 JVM 中运行。

要将 Spark 应用程序提交到正在运行的 YARN 集群，请设置 setMaster("yarn") 并可选择将 deploy-mode 属性设置为客户端或集群。

有关这些参数的更多详细信息，请参阅this。

另外，如果您想通过代码而不是命令行提交应用程序，请参考this post。