0570-如何在CDH集群上部署Python3.6.1环境及运行Pyspark作业

Fayson

发布于 2019-04-29 16:12:18

3.1K0

发布于 2019-04-29 16:12:18

文章被收录于专栏：Hadoop实操

温馨提示：如果使用电脑查看图片不清晰，可以使用手机打开文章单击文中的图片放大查看高清原图。
Fayson的github： https://github.com/fayson/cdhproject
提示：代码块部分可以左右滑动查看噢

文档编写目的

当前有很多工具辅助大数据分析，但最受环境的就是Python。Python简单易用，语言有着直观的语法并且提供强大的科学计算和集群学习库。借着最近人工智能，深度学习的兴起，Python成为时下最火的语言，已经超越了Java和C，并且纳入了国家计算机等级考试。本篇文章主要讲述如何在CDH集群基于Anaconda安装包部署Python3.6.1的运行环境，并使用PySpark作业验证Python3环境的可行性。

测试环境

1.CM和CDH版本为5.16.1

2.集群已启用Kerberos

3.Spark 2.3.0.cloudera4

基于Anaconda部署Python3

注意在集群的每个节点都要部署python3.6.1，并且部署目录要相同，因为CDH自带了python2，所以如果需要将python替换成python3，需要在CM界面上修改PYSPARK_PYTHON的路径，下面会有说明。

下载anaconda安装包，这里选用的版本是Anaconda3-4.4.0-Linux-x86_64.sh，下载地址：https://repo.continuum.io/archive/Anaconda3-4.4.0-Linux-x86_64.sh，这个版本对应的Python版本是3.6.1。

1.进入到安装包目录，执行命令：bash Anaconda3-4.4.0-Linux-x86_64.sh

2.下一步输入回车键

3.在以下界面输入”yes”

4.下一步设置安装路径，我们这里安装在/opt/cloudera/anaconda3目录下

如果提示“tar (child): bzip2: Cannot exec: No such file or directory”，需要先安装bzip2。

[root@ip-172-31-13-38 ~]$ sudo yum -y install bzip2

5.安装完后，提示设置anaconda的PATH路径，这里需要设置全局路径，因为要确保pyspark任务提交过来之后可以使用python3，所以输入“no”，重新设置PATH

6.设置全局的anaconda3的PATH

[root@ip-172-31-13-38 ec2-user]# echo "export PATH=/opt/cloudera/anaconda3/bin:$PATH" >> /etc/profile
[root@ip-172-31-13-38 ec2-user]# source /etc/profile
[root@ip-172-31-13-38 ec2-user]# env |grep PATH
PATH=/usr/local/anaconda3/bin:/sbin:/bin:/usr/sbin:/usr/bin
[root@ip-172-31-13-38 ec2-user]#

7.执行如下命令,验证Python版本

在CM配置Spark2的Python环境

1.通过export设置python命令的安装路径：

export PYSPARK_PYTHON=/opt/cloudera/anaconda3/bin/python
export PYSPARK_DRIVER_PYTHON=/opt/cloudera/anaconda3/bin/python

修改完成后，回到CM主页根据提示重启相关服务。

pyspark命令测试

1.获取kerberos凭证

2.使用Pyspark2命令测试

x = sc.parallelize([1,2,3])
y = x.flatMap(lambda x: (x, 100*x, x**2))
print(x.collect())
print(y.collect())

提交一个Pyspark作业

这个demo主要使用spark2-submit提交pyspark job，模拟从hdfs中读取数据，并转换成DateFrame，然后注册为临时表并执行SQL条件查询，将查询结果输出到hdfs中。

1.将测试数据上传至hdfs目录/tmp/examples/

[root@ip-172-31-13-38 ~]# klist
[root@ip-172-31-13-38 ~]# hadoop fs -mkdir /tmp/examples/
[root@ip-172-31-13-38 ~]# hadoop fs -put people.txt /tmp/examples
[root@ip-172-31-13-38 ~]# hadoop fs -cat /tmp/examples/people.txt

2.在集群的一个部署了Spark2 Gateway角色和Python3环境的节点上编写PySparkTest2HDFS.py程序内容如下：

# 初始化sqlContext
from pyspark import SparkConf,SparkContext
from pyspark.sql import SQLContext, Row
conf=(SparkConf().setAppName('PySparkTest2HDFS'))
sc=SparkContext(conf=conf)
sqlContext = SQLContext(sc)

# 加载文本文件并转换成Row.
lines = sc.textFile("/tmp/examples/people.txt")
parts = lines.map(lambda l: l.split(","))
people = parts.map(lambda p: Row(name=p[0], age=int(p[1])))

# 将DataFrame注册为table.
schemaPeople = sqlContext.createDataFrame(people)
schemaPeople.registerTempTable("people")

# 执行sql查询，查下条件年龄在13岁到19岁之间
teenagers = sqlContext.sql("SELECT name,age FROM people WHERE age >= 13 AND age <= 19")

# 将查询结果保存至hdfs中
teenagers.write.save("/tmp/examples/teenagers")

3.使用spark2-submit命令向集群提交PySpark作业

[root@ip-172-31-13-38 ~]# spark2-submit PySparkTest2HDFS.py

4.作业执行成功

查看Spark2的HistoryServer

通过以上信息，可以看到作业执行成功。

5. 查看生成的文件，如下图：

因为生成的是parquet文件，它是二进制文件，无法直接使用命令查看，所以我们可以在pyspark上验证文件内容是否正确.

我们上面使用spark2-submit提交的任务使用sql查询条件是3到4岁，可以看到在pyspark2上查询的数据是在这个区间的数据

parquetFile = sqlContext.read.parquet("/tmp/examples/teenagers")
parquetFile.registerTempTable("parquetTable")
teenagers = sqlContext.sql("select * from parquetTable")
teenagers.show()