如何运行 Spark Java 程序

Question

我为 Spark 编写了一个 Java 程序。但是如何从 Unix 命令行运行和编译它。编译运行时是否必须包含任何 jar 文件

Answer 1

结合官方Quick Start Guide和Launching Spark on YARN的步骤我们得到：

我们将创建一个非常简单的 Spark 应用程序 SimpleApp.java：

/*** SimpleApp.java ***/
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.Function;

public class SimpleApp {
  public static void main(String[] args) {
    String logFile = "$YOUR_SPARK_HOME/README.md"; // Should be some file on your system
    JavaSparkContext sc = new JavaSparkContext("local", "Simple App",
      "$YOUR_SPARK_HOME", new String[]{"target/simple-project-1.0.jar"});
    JavaRDD<String> logData = sc.textFile(logFile).cache();

    long numAs = logData.filter(new Function<String, Boolean>() {
      public Boolean call(String s) { return s.contains("a"); }
    }).count();

    long numBs = logData.filter(new Function<String, Boolean>() {
      public Boolean call(String s) { return s.contains("b"); }
    }).count();

    System.out.println("Lines with a: " + numAs + ", lines with b: " + numBs);
  }
}

这个程序只计算文本文件中包含“a”的行数和包含“b”的行数。请注意，您需要将 $YOUR_SPARK_HOME 替换为安装 Spark 的位置。与 Scala 示例一样，我们初始化了一个 SparkContext，尽管我们使用特殊的 JavaSparkContext 类来获得一个对 Java 友好的类。我们还创建 RDD（由 JavaRDD 表示）并在它们上运行转换。最后，我们通过创建扩展 spark.api.java.function.Function 的类将函数传递给 Spark。 Java 编程指南更详细地描述了这些差异。

为了构建程序，我们还编写了一个 Maven pom.xml 文件，将 Spark 列为依赖项。请注意，Spark 工件标有 Scala 版本。

<project>
  <groupId>edu.berkeley</groupId>
  <artifactId>simple-project</artifactId>
  <modelVersion>4.0.0</modelVersion>
  <name>Simple Project</name>
  <packaging>jar</packaging>
  <version>1.0</version>
  <repositories>
    <repository>
      <id>Akka repository</id>
      <url>http://repo.akka.io/releases</url>
    </repository>
  </repositories>
  <dependencies>
    <dependency> <!-- Spark dependency -->
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_2.10</artifactId>
      <version>0.9.0-incubating</version>
    </dependency>
  </dependencies>
</project>

如果您还希望从 Hadoop 的 HDFS 读取数据，您还需要为您的 HDFS 版本添加对 hadoop-client 的依赖：

<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-client</artifactId>
  <version>...</version>
</dependency>

我们根据规范的 Maven 目录结构对这些文件进行布局：

$ find .
./pom.xml
./src
./src/main
./src/main/java
./src/main/java/SimpleApp.java

现在，我们可以使用 Maven 执行应用程序了：

$ mvn package
$ mvn exec:java -Dexec.mainClass="SimpleApp"
...
Lines with a: 46, Lines with b: 23

然后按照Launching Spark on YARN的步骤：

构建支持 YARN 的程序集 JAR

我们需要一个整合的 Spark JAR（它捆绑了所有必需的依赖项）来在 YARN 集群上运行 Spark 作业。这可以通过设置 Hadoop 版本和 SPARK_YARN 环境变量来构建，如下所示：

SPARK_HADOOP_VERSION=2.0.5-alpha SPARK_YARN=true sbt/sbt assembly

组装后的 JAR 将如下所示：./assembly/target/scala-2.10/spark-assembly_0.9.0-incubating-hadoop2.0.5.jar。

构建过程现在还支持新的 YARN 版本 (2.2.x)。见下文。

准备工作

• 构建启用 YARN 的程序集（见上文）。
• 组装好的jar可以安装到HDFS或者本地使用。
• 您的应用程序代码必须打包到单独的 JAR 文件中。

如果您想测试 YARN 部署模式，可以使用当前的 Spark 示例。通过运行可以生成一个 spark-examples_2.10-0.9.0-incubating 文件：

sbt/sbt assembly 

注意：由于您正在阅读的文档是针对 Spark 0.9.0-incubating 版本的，因此我们假设您已下载 Spark 0.9.0-incubating 版本或将其从源代码控制中检出.如果你使用的是不同版本的Spark，sbt package命令生成的jar中的版本号显然会有所不同。

配置

Yarn 上的 Spark 的大多数配置与其他部署相同。有关这些的更多信息，请参阅配置页面。这些是特定于 YARN 上的 SPARK 的配置。

环境变量：

• SPARK_YARN_USER_ENV，将环境变量添加到在 YARN 上启动的 Spark 进程。这可以是逗号分隔的环境变量列表，例如

SPARK_YARN_USER_ENV="JAVA_HOME=/jdk64,FOO=bar"

系统属性：

• spark.yarn.applicationMaster.waitTries，属性设置 ApplicationMaster 等待 spark master 的次数，以及它等待 Spark Context 初始化的尝试次数。默认值为 10。
• spark.yarn.submit.file.replication，应用程序上传到 HDFS 的文件的 HDFS 复制级别。其中包括 spark jar、app jar 和任何分布式缓存文件/档案等内容。
• spark.yarn.preserve.staging.files，设置为 true 以在作业结束时保留暂存文件（spark jar、app jar、分布式缓存文件）而不是删除它们。
• spark.yarn.scheduler.heartbeat.interval-ms，Spark 应用程序 master 心跳进入 YARN ResourceManager 的时间间隔（毫秒）。默认为 5 秒。
• spark.yarn.max.worker.failures，应用程序失败前的最大工作失败次数。默认值为请求的工作人员数量乘以 2，最少为 3。

在 YARN 上启动 Spark

确保 HADOOP_CONF_DIR 或 YARN_CONF_DIR 指向包含 hadoop 集群（客户端）配置文件的目录。这将用于连接到集群、写入 dfs 并将作业提交给资源管理器。

有两种调度模式可用于在 YARN 上启动 Spark 应用程序。

YARN 客户端以 yarn-standalone 模式启动 spark 应用程序。

启动YARN Client的命令如下：

SPARK_JAR=<SPARK_ASSEMBLY_JAR_FILE> ./bin/spark-class org.apache.spark.deploy.yarn.Client \
  --jar <YOUR_APP_JAR_FILE> \
  --class <APP_MAIN_CLASS> \
  --args <APP_MAIN_ARGUMENTS> \
  --num-workers <NUMBER_OF_WORKER_MACHINES> \
  --master-class <ApplicationMaster_CLASS>
  --master-memory <MEMORY_FOR_MASTER> \
  --worker-memory <MEMORY_PER_WORKER> \
  --worker-cores <CORES_PER_WORKER> \
  --name <application_name> \
  --queue <queue_name> \
  --addJars <any_local_files_used_in_SparkContext.addJar> \
  --files <files_for_distributed_cache> \
  --archives <archives_for_distributed_cache>

例如：

# Build the Spark assembly JAR and the Spark examples JAR
$ SPARK_HADOOP_VERSION=2.0.5-alpha SPARK_YARN=true sbt/sbt assembly

# Configure logging
$ cp conf/log4j.properties.template conf/log4j.properties

# Submit Spark's ApplicationMaster to YARN's ResourceManager, and instruct Spark to run the SparkPi example
$ SPARK_JAR=./assembly/target/scala-2.10/spark-assembly-0.9.0-incubating-hadoop2.0.5-alpha.jar \
    ./bin/spark-class org.apache.spark.deploy.yarn.Client \
      --jar examples/target/scala-2.10/spark-examples-assembly-0.9.0-incubating.jar \
      --class org.apache.spark.examples.SparkPi \
      --args yarn-standalone \
      --num-workers 3 \
      --master-memory 4g \
      --worker-memory 2g \
      --worker-cores 1

# Examine the output (replace $YARN_APP_ID in the following with the "application identifier" output by the previous command)
# (Note: YARN_APP_LOGS_DIR is usually /tmp/logs or $HADOOP_HOME/logs/userlogs depending on the Hadoop version.)
$ cat $YARN_APP_LOGS_DIR/$YARN_APP_ID/container*_000001/stdout
Pi is roughly 3.13794

上面启动了一个启动默认Application Master的YARN客户端程序。然后 SparkPi 将作为 Application Master 的子线程运行，YARN Client 将定期轮询 Application Master 以获取状态更新并将其显示在控制台中。一旦您的应用程序完成运行，客户端将退出。

使用此模式，您的应用程序实际上是在运行 Application Master 的远程计算机上运行的。因此，涉及本地交互的应用程序将无法正常工作，例如火花壳。

Answer 2

几天前我有同样的问题，昨天设法解决了。
这就是我所做的：

1. 下载sbt并解压解压:http://www.scala-sbt.org/download.html
2. 我已经下载了适用于 Hadoop 2 的 Spark Prebuild 包，解压并解压：http://www.apache.org/dyn/closer.cgi/spark/spark-1.0.2/spark-1.0.2-bin-hadoop2.tgz
3. 我已经创建了独立的应用程序 SimpleApp.scala，如：http://spark.apache.org/docs/latest/quick-start.html#standalone-applications 中所述，具有正确的 simple.sbt 文件（刚刚从描述中复制）和正确的目录布局
4. 确保 PATH 中有 sbt。使用您的应用程序转到目录并使用 sbt package 构建您的包
5. 使用SPARK_HOME_DIR/sbin/spark_master.sh启动Spark Server
6. 转到localhost:8080 并确保您的服务器正在运行。从 URL 复制链接（来自服务器描述，而不是本地主机。它应该是端口 7077 或类似的东西）
7. 使用SPARK_HOME_DIR/bin/spark-class org.apache.spark.deploy.worker.Worker spark://IP:PORT启动Worker，其中IP:PORT是6中复制的URL
8. 将您的应用程序部署到服务器：SPARK_HOME_DIR/bin/spark-submit --class "SimpleApp" --master URL target/scala-2.10/simple-project_2.10-1.0.jar

这对我有用，希望对你有所帮助。
帕维尔

Answer 3

除了所选答案之外，如果您想连接到外部独立 Spark 实例：

SparkConf conf =
new SparkConf()
     .setAppName("Simple Application")
     .setMaster("spark://10.3.50.139:7077");

JavaSparkContext sc = new JavaSparkContext(conf);

根据 Spark 的运行位置，您可以在此处找到更多“主”配置：http://spark.apache.org/docs/latest/submitting-applications.html#master-urls

Answer 4

这个答案适用于 Spark 2.3。如果你想在本地测试你的 Spark 应用程序，即没有 Hadoop 集群的先决条件，甚至不需要启动任何独立的 Spark 服务，你可以这样做：

JavaSparkContext jsc = new JavaSparkContext(new SparkConf().setAppName("Simple App"));

然后，在本地运行您的应用程序：

$SPARK_HOME/bin/spark-submit --class SimpleApp --master local target/scala-2.10/simple-project_2.10-1.0.jar

为此，您只需将 Spark tar 文件解压缩到 $SPARK_HOME，并将 $SPARK_HOME 设置到 Spark 用户的 .profile 中