YARN 集群上的 PySpark 分布式处理

Question

我让Spark在Cloudera CDH5.3集群上运行，使用YARN作为资源管理器。我正在用Python（PySpark）开发Spark应用程序。

我可以提交作业并且它们成功运行，但是它们似乎从未在多台计算机（我提交的本地计算机）上运行。

我已经尝试了各种选项，例如将 --deploy-mode 设置为集群，将 --master 设置为 yarn-client 和 yarn-cluster，但它似乎从未在多台服务器上运行过。

我可以通过传递类似 --master local[8] 的东西让它在多个内核上运行，但这显然不会将处理分布在多个节点上。

我有一个非常简单的Python脚本处理来自HDFS的数据，如下所示：

import simplejson as json
from pyspark import SparkContext
sc = SparkContext("", "Joe Counter")
rrd = sc.textFile("hdfs:///tmp/twitter/json/data/")
data = rrd.map(lambda line: json.loads(line))
joes = data.filter(lambda tweet: "Joe" in tweet.get("text",""))
print joes.count()

我正在运行一个提交命令，例如：

spark-submit atest.py --deploy-mode client --master yarn-client

如何确保作业在群集中并行运行？

Answer 1

你能交换命令的参数吗？火花-提交--部署模式客户端--主纱线客户端 atest.py

如果您看到该命令的帮助文本：

火花提交

Usage: spark-submit [options] <app jar | python file>

Answer 2

我相信

@MrChristine是正确的 - 您指定的选项标志正在传递给您的python脚本，而不是用于Spark提交。此外，您需要指定--executor-cores和--num-executors因为默认情况下它将在单个内核上运行并使用两个执行器。

Answer 3

python脚本不在集群模式下运行是不正确的。我不确定以前的版本，但这是在 Hortonworks 集群上的 Spark 2.2 版本中执行的。

命令 ： 火花-提交 --主纱线 --执行器数 10 --执行器核心数 1 --驱动内存 5g/pyspark-example.py

蟒蛇代码：

from pyspark import SparkConf, SparkContext
from pyspark.sql import SQLContext
conf = (SparkConf()
         .setMaster("yarn")
         .setAppName("retrieve data"))
sc = SparkContext(conf = conf)
sqlContext = SQLContext(sc)
parquetFile = sqlContext.read.parquet("/<hdfs-path>/*.parquet")
parquetFile.createOrReplaceTempView("temp")
df1 = sqlContext.sql("select * from temp limit 5")
df1.show()
df1.write.save('/<hdfs-path>/test.csv', format='csv', mode='append')
sc.stop()

输出：它很大，所以我不粘贴。但它运行完美。

Answer 4

似乎 PySpark 不能使用 Spark/YARN 在分布式模式下运行 - 您需要将独立的 Spark 与 Spark Master 服务器一起使用。在这种情况下，我的 PySpark 脚本在整个集群中运行得很好，每个核心/节点都有一个 Python 进程。