通过 Yarn 集群与客户端提交时的 Spark 输出

Question

我是 Spark 的新手，刚刚在我的集群上运行它（运行社区版 MapR 的 9 节点集群上的 Spark 2.0.1）。我通过

提交字数示例

./bin/spark-submit --master yarn --jars ~/hadoopPERMA/jars/hadoop-lzo-0.4.21-SNAPSHOT.jar examples/src/main/python/wordcount.py ./README.md

并得到以下输出

17/04/07 13:21:34 WARN Client: Neither spark.yarn.jars nor spark.yarn.archive is set, falling back to uploading libraries under SPARK_HOME.
: 68
help: 1
when: 1
Hadoop: 3
...

看起来一切正常。当我添加 --deploy-mode cluster 时，我得到以下输出：

17/04/07 13:23:52 WARN Client: Neither spark.yarn.jars nor spark.yarn.archive is set, falling back to uploading libraries under SPARK_HOME.

所以没有错误但我没有看到字数统计结果。我错过了什么？我在我的历史服务器中看到了该作业，它说它已成功完成。我还检查了 DFS 中的用户目录，但除了这个空目录之外没有写入新文件：/user/myuser/.sparkStaging

代码（Spark 附带的 wordcount.py 示例）：

from __future__ import print_function
import sys
from operator import add
from pyspark.sql import SparkSession


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: wordcount <file>", file=sys.stderr)
        exit(-1)

    spark = SparkSession\
        .builder\
        .appName("PythonWordCount")\
        .getOrCreate()

    lines = spark.read.text(sys.argv[1]).rdd.map(lambda r: r[0])
    counts = lines.flatMap(lambda x: x.split(' ')) \
              .map(lambda x: (x, 1)) \
              .reduceByKey(add)
    output = counts.collect()
    for (word, count) in output:
        print("%s: %i" % (word, count))

    spark.stop()

Answer 1

您的输出未打印的原因是：

当您在 spark-client 模式下运行时，您启动作业的节点是 DRIVER，当您收集结果时，它会在该节点上收集并打印。

在 yarn-cluster 模式下，您的驱动程序是某个其他节点，而不是您启动作业的节点。因此，当您调用 .collect 函数时，结果会在该节点上收集并打印在该节点上。您可以在驱动程序的 sys-out 中找到正在打印的结果。 更好的方法是将输出写入 HDFS 中的某处。

您的 spark.yarn.jars 警告的原因是：

为了运行 Spark 作业，yarn 需要在集群的所有节点上提供一些二进制文件，如果这些二进制文件不可用，那么作为作业准备的一部分，Spark 将创建一个包含 $SPARK_HOME/jars 下所有 jar 的 zip 文件并将其上传到分布式缓存。

解决这个问题：

默认情况下，YARN 上的 Spark 将使用本地安装的 Spark jar，但 Spark jar 也可以位于 HDFS 上的全局可读（chmod 777）位置。这允许 YARN 将其缓存在节点上，这样就不需要在每次应用程序运行时分发它。例如，要指向 HDFS 上的 jars，请将 spark.yarn.jars 设置为 hdfs:///some/path。

放置你的 jars 后运行你的代码：

./bin/spark-submit --master yarn --jars ~/hadoopPERMA/jars/hadoop-lzo-0.4.21-SNAPSHOT.jar examples/src/main/python/wordcount.py ./README.md --conf spark.yarn.jars="hdfs:///some/path"

来源：http://spark.apache.org/docs/latest/running-on-yarn.html