无法使用java初始化spark上下文

Question

我正在尝试使用spark一个简单的工作计数程序，但是当我尝试初始化spark上下文时它失败了。 下面是我的代码

conf = new SparkConf(true).
                setAppName("WordCount").
                setMaster("spark://192.168.0.104:7077");

        sc = new JavaSparkContext(conf);

现在我想澄清一些事情，我正在使用 Spark 版本 2.1.1 ，我的 java 代码在 Windows 10 上，我的服务器在 VM 盒上运行。 我在虚拟机中禁用了防火墙，并且可以从 Windows 访问 URL http://192.168.0.104:8080/。

但是我在运行代码时低于堆栈跟踪

17/08/06 18:44:15 INFO SparkUI: Bound SparkUI to 0.0.0.0, and started at http://192.168.0.103:4040
17/08/06 18:44:15 INFO StandaloneAppClient$ClientEndpoint: Connecting to master spark://192.168.0.104:7077...
17/08/06 18:44:15 INFO TransportClientFactory: Successfully created connection to /192.168.0.104:7077 after 41 ms (0 ms spent in bootstraps)
17/08/06 18:44:15 WARN StandaloneAppClient$ClientEndpoint: Failed to connect to master 192.168.0.104:7077
org.apache.spark.SparkException: Exception thrown in awaitResult
    at org.apache.spark.rpc.RpcTimeout$$anonfun$1.applyOrElse(RpcTimeout.scala:77)
    at org.apache.spark.rpc.RpcTimeout$$anonfun$1.applyOrElse(RpcTimeout.scala:75)
    at scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:33)
    at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:59)
    at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:59)

有人可以帮忙吗？

Answer 1

有点晚了，但对于现在遇到这种情况的人来说：这可能是由于 Spark Core 或 Spark SQL 使用的 Maven 版本与服务器上使用的 Spark 版本不兼容所致。目前 Spark 2.4.4 似乎与以下 Maven 设置兼容：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.11</artifactId>
    <version>2.3.4</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.11</artifactId>
    <version>2.3.4</version>
</dependency>

可以通过查看Spark Master节点日志来诊断不兼容问题。他们应该提到类似

spark local class incompatible stream classdesc serialversionuid

的内容

我希望这对某些人仍然有用！

Answer 2

您需要导入正确的Spark导入java文件以及添加相关的maven依赖项。这是使用 JDK 17 的示例。

Java 文件：

import org.apache.spark.*;
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.JavaSparkContext;
import java.io.Serializable;


public static void test(String[] input){
        SparkConf conf = new SparkConf().setAppName("app").setMaster("local[*]");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaRDD<Foo> rdd = sc.parallelize(input)
            .map(p->{
            return new Foo(p);
        });
        rdd.collect().forEach(System.out::println);
}

将以下 Maven 依赖项添加到 Maven pom.xml：

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_2.13</artifactId>
      <version>3.5.1</version>
      <type>jar</type>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-client</artifactId>
      <version>3.4.1</version>
      <type>jar</type>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-mapreduce-client-core</artifactId>
      <version>3.4.1</version>
      <type>jar</type>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-hdfs</artifactId>
      <version>3.4.1</version>
      <type>jar</type>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-core</artifactId>
      <version>1.2.1</version>
      <type>jar</type>
    </dependency>

Answer 3

您需要将一些 Spark 类导入到您的程序中。添加以下行：

import org.apache.spark.api.java.JavaSparkContext
import org.apache.spark.api.java.JavaRDD
import org.apache.spark.SparkConf

SparkConf conf = new SparkConf().setAppName("WordCount").setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);