maven - 客户端 yarn 作业的 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”吗？是否存在依赖性问题？如果是，普通客户端和 yarn-client 在依赖方面有什么不同？ 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 设置为 local[*] 时，您的 spark 应用程序在您机器的单个 JVM 中运行。

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

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

此外，如果您想从代码而不是命令行提交申请，请引用 this post .