如何在DataWorks上运行PySpark作业_大数据开发治理平台 DataWorks-阿里云帮助中心

PySpark可以直接调用Python的API运行Spark作业，本文为您介绍如何在DataWorks上运行PySpark作业。

前提条件

在DataWorks运行PySpark作业时，需要创建Spark节点，使用spark-submit命令提交作业。

创建Spark节点，详情请参见创建EMR Spark节点。

提交作业前需确认以下配置项填写正确。

spark-submit --master yarn \
--deploy-mode cluster \
--conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./PYTHONENV/python3/bin/python3.7 \
--conf spark.executorEnv.PYTHONPATH=. \
--conf spark.yarn.appMasterEnv.PYTHONPATH=. \
--archives hdfs://hdfs-cluster/tmp/pyspark/python3.7.zip#PYTHONENV \
--name TestPySpark \
hdfs://hdfs-cluster/tmp/pyspark/pyspark_test.py

说明 DataWorks EMR资源的使用上限为50M，您需根据添加依赖的Python包大小选择上传方式：

大于50M时，直接上传至HDFS或OSS。
小于50M时，您可选择在DataWorks上传。

操作步骤

准备运行Python程序需要的虚拟环境。

制作Docker镜像。

在安装了Docker环境的宿主机上新建一个Dockerfile文件，Python3代码示例如下。

FROM centos:7.6.1810
RUN set -ex \
# 预安装所需组件。
&& yum install -y wget tar libffi-devel zlib-devel bzip2-devel openssl-devel ncurses-devel sqlite-devel readline-devel tk-devel gcc make initscripts zip\
&& wget https://www.python.org/ftp/python/3.7.0/Python-3.7.0.tgz \
&& tar -zxvf Python-3.7.0.tgz \
&& cd Python-3.7.0 \
&& ./configure prefix=/usr/local/python3 \
&& make \
&& make install \
&& make clean \
&& rm -rf /Python-3.7.0* \
&& yum install -y epel-release \
&& yum install -y python-pip
# 设置默认为python3。
RUN set -ex \
# 备份旧版本python。
&& mv /usr/bin/python /usr/bin/python27 \
&& mv /usr/bin/pip /usr/bin/pip-python27 \
# 配置默认为python3。
&& ln -s /usr/local/python3/bin/python3.7 /usr/bin/python \
&& ln -s /usr/local/python3/bin/pip3 /usr/bin/pip
# 修复因修改python版本导致yum失效问题。
RUN set -ex \
&& sed -i "s#/usr/bin/python#/usr/bin/python27#" /usr/bin/yum \
&& sed -i "s#/usr/bin/python#/usr/bin/python27#" /usr/libexec/urlgrabber-ext-down \
&& yum install -y deltarpm
# 更新pip版本。
RUN pip install --upgrade pip

构建镜像并运行容器。

docker build -t python-centos:3.7 .
docker run -itd --name python3.7 python-centos:3.7

进入安装容器所需的Python依赖库并打包Python环境。

docker exec -it  python3.7 bash

pip install [所需依赖库]
# vi requirements.txt
# pip install -r requirements.txt
# numpy
# pandas

cd /usr/local/
zip -r python3.7.zip python3/

拷贝容器中的Python环境到宿主机，上传至HDFS中。

# 在宿主机运行命令将虚拟环境拷贝到宿主机。
docker cp python3.7:/usr/local/python3.7.zip .

# 上传至HDFS中。
hdfs dfs -copyFromLocal python3.7.zip /tmp/pyspark

测试并上传Python代码。

按照如下方法测试pyspark_test.py。

# -*- coding: utf-8 -*-

import os
from pyspark.sql import SparkSession

def noop(x):
    import socket
    import sys
    host = socket.gethostname() + ' '.join(sys.path) + ' '.join(os.environ)
    print('host: ' + host)
    print('PYTHONPATH: ' + os.environ['PYTHONPATH'])
    print('PWD: ' + os.environ['PWD'])
    print(os.listdir('.'))
    return host


if __name__ == '__main__':

    # 本地调试时添加，MaxCompute运行时需要删除，否则会报错。
    # .master("local[4]") \
    spark = SparkSession \
        .builder \
        .appName("test_pyspark") \
        .enableHiveSupport() \
        .getOrCreate()

    sc = spark.sparkContext
    # 验证系统当前环境变量。
    rdd = sc.parallelize(range(10), 2)
    hosts = rdd.map(noop).distinct().collect()
    print(hosts)

    # 验证UDF。
    # https://docs.databricks.com/spark/latest/spark-sql/udf-python.html#
    # spark.udf.register("udf_squared", udf_squared)
    # spark.udf.register("udf_numpy", udf_numpy)

    tableName = "store"
    df = spark.sql("""select count(*) from %s """ % tableName)
    print("rdf count, %s\n" % df.count())
    df.show()

上传Python代码至HDSF中。

hdfs dfs -copyFromLocal pyspark_test.py /tmp/pyspark

在DataWorks中通过spark-submit命令提交作业。

spark-submit --master yarn \
--deploy-mode cluster \
--conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./PYTHONENV/python3/bin/python3.7 \
--conf spark.executorEnv.PYTHONPATH=. \
--conf spark.yarn.appMasterEnv.PYTHONPATH=. \
--conf spark.yarn.appMasterEnv.JOBOWNER=LiuYuQuan \
--archives hdfs://hdfs-cluster/tmp/pyspark/python3.7.zip#PYTHONENV \
## --py-files hdfs://hdfs-cluster/tmp/pyspark/mc_pyspark-0.1.0-py3-none-any.zip \
--driver-memory 4g \
--driver-cores 1 \
--executor-memory 4g \
--executor-cores 1 \
--num-executors 3 \
--name TestPySpark \
hdfs://hdfs-cluster/tmp/pyspark/pyspark_test.py