使用spark-submit时python脚本在哪里执行？

Question

Python : 3.7.3
OS: CentOS 7
Spark: 2.2.0
In Cloudera
YARN : 2.6.0-cdh5.10.2

嗨，我尝试用 python 脚本和 pyspark[=45] 执行 Apache Spark =]，但我不明白它是如何工作的。我尝试在执行 spark-submit 时发送整个 conda 环境，在客户端模式 yarn 中使用参数 --archives。但问题是，主要 python 脚本 运行 在哪里，因为我需要指定我的共享 conda 环境的位置才能无错误地执行，因为在我尝试执行的主机中 spark-submit 我还没有安装依赖项，我不想安装它。

我使用这个特性来打包环境https://conda.github.io/conda-pack/spark.html，我需要在映射之外导入依赖项（因为在映射中，yarn 运送了依赖项并且执行者很好地导入了这个依赖项）。

有没有一种方法可以在不在主机上打开和使用的情况下使用附带的环境执行主要 python 脚本？

我的环境是：

PYSPARK_DRIVER_PYTHON=./enviroment/bin/python
PYSPARK_PYTHON=./enviroment/bin/python

其中环境是 yarn

附带的依赖项的符号 link

--archives ~/dependencies.tar.gz#enviroment

并配置执行器

--conf spark.yarn.appMasterEnv.PYSPARK_DRIVER_PYTHON=./environment/bin/python

所以最后的命令是

PYSPARK_DRIVER_PYTHON=./enviroment/bin/python \
PYSPARK_PYTHON=./environment/bin/python \
spark-submit \
--conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./environment/bin/python \
--master yarn --deploy-mode client \
--archives enviroment/dependencies.tar.gz#enviroment \
cluster-import-check.py

我的密码是

# coding=utf8
from pyspark import SparkConf
from pyspark import SparkContext

import sys
import numpy

def check_import(version, info=None):
    print("=====VERSION : ", version)
    if info and type(info) == list and len(info) != 0:
        for i in info:
            print("=====INFO EXTRA : ", i)

def python_import(x):
    import sys
    print("\n===PYTHON")
    check_import(sys.version, [sys.executable])

def numpy_import(x):
    import numpy
    print("\n===NUMPY")
    check_import(numpy.__version__, [numpy.__file__])

def printInfo(object):
    print("=====NAME : ", object.__name__)
    if object.__name__ == 'sys':
        print("=====VERSION", object.version)
        print("=====LOCATED IN", object.executable)
    else:
        print("=====VERSION : ", object.__version__)
        print("=====LOCATED IN : ", object.__file__)

    if object.__name__ == 'elasticsearch':
        es = elasticsearch.Elasticsearch(['172.22.248.206:9201'])
        print("=====MORE INFO : ", es.info())

def init_spark():
    conf = SparkConf()
    conf.setAppName("imports-checking")

    sc = SparkContext(conf=conf).getOrCreate()

    return conf, sc


def main():
    conf, sc = init_spark()
    print(sc.getConf().getAll())

    print(sc.parallelize([0]).map(lambda x: python_import(x)).collect())

    sc.stop()

if __name__ == '__main__':
    printInfo(sys)
    printInfo(numpy)
    main()

并且一个错误是no module named numpy或者定位的python是另一个错误，因为在集群中还有另一个版本的python，但我想使用由集群上的纱线。

Answer 1

我最近了解 PysPark 和 Yarn 的工作流程，答案是如果你想 运行 在客户端模式下你需要安装（在你执行 spark-submit 的主机上）所有导入的库在函数映射之外。另一方面，如果您想 运行 在集群模式下，您只需要在 spark-submit 命令中使用选项 --archives 运送库。

1. 本地

执行时，它是在本地完成的，必须在执行 spark-submit

时配置 PYSPARK_DRIVER_PYTHON

PYSPARK_DRIVER_PYTHON=./dependencies/bin/python spark-submit --master local --py-files cognition.zip MainScript.py

2. 纱线客户端

在执行spark-submit命令的主机上执行。必须添加环境变量

PYSPARK_DRIVER_PYTHON=./dependencies/bin/python spark-submit --conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./dependencies/bin/python --master yarn --deploy-mode client --archives dependencies.tar.gz#dependencies MainScript.py

4. 纱线集群

执行在其创建的yarn容器内部进行，在集群任意节点执行

spark-submit --conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./dependencies/bin/python --master yarn --deploy-mode cluster --archives dependencies.tar.gz#dependencies MainScript.py