如何在CDH集群上部署Python3运行环境及运行Python作业
如何在CDH集群上部署Python3运行环境及运行Python作业
温馨提示：要看高清无码套图，请使用手机打开并单击图片放大查看。 Fayson的github：https://github.com/fayson/cdhproject
1.文档编写目的
当前有很多工具辅助大数据分析，但最受欢迎的就是Python。Python简单易用，语言有着直观的语法并且提供强大的科学计算和集群学习库。借着最近人工智能，深度学习的兴起，Python成为时下最火的语言，已经超越了Java和C，并且纳入了国家计算机等级考试。本篇文章主要讲述如何在CDH集群基于Anaconda部署Python3的运行环境，并使用示例说明使用pyspark运行Python作业。
- 测试环境
1.CM和CDH版本为5.11.2
2.采用sudo权限的ec2-user用户操作
3.集群已启用Kerberos
- 前置条件
1.Spark On Yarn模式
2.基于Anaconda部署Python3
注意在每个worker节点都要部署python3，并且部署目录要相同，因为CDH自带了python2，所以如果需要将python替换成python3，需要在CM界面上修改PYSPARK_PYTHON的路径，下面会有说明。
下载anaconda包，这里选用的版本是Anaconda3-4.2.0-Linux-x86_64，下载地址：
https://repo.continuum.io/archive/Anaconda3-4.2.0-Linux-x86_64.sh
这个版本对应的Python版本是3.5.2，版本需要注意的是PySpark does not work with Python 3.6.0，SPARK-19019
https://issues.apache.org/jira/browse/SPARK-19019
所以我们这里装Python3.5.2版本。
1.进入到安装包目录，执行命令：bashAnaconda3-4.2.0-Linux-x86_64.sh
2.下一步输入回车键
3.在以下界面输入”yes”
4.下一步设置安装路径，我们这里安装在/usr/local/anaconda3目录下
如果提示“tar(child): bzip2: Cannot exec: No such file or directory”，需要先安装bzip2。
ec2-user@ip-172-31-21-45 ~$ sudo yum -y install bzip2
5.安装完后，提示设置anaconda的PATH路径，这里需要设置全局路径，因为要确保pyspark任务提交过来之后可以使用python3，所以输入“no”，重新设置PATH
6.设置全局的anaconda3的PATH
[root@ip-172-31-21-45 ec2-user]# echo "export PATH=/usr/local/anaconda3/bin:$PATH" >> /etc/profile
[root@ip-172-31-21-45 ec2-user]# source /etc/profile
[root@ip-172-31-21-45 ec2-user]# env |grep PATH
PATH=/usr/local/anaconda3/bin:/sbin:/bin:/usr/sbin:/usr/bin
[root@ip-172-31-21-45 ec2-user]# 
7.使用conda创建python3环境并激活
执行命令：
[root@ip-172-31-21-45 ec2-user]# conda create --unknown --offline -n py3 python=3.5
注意：这里创建python3环境时使用了离线模式，即--offline参数，以及—unknown，这种方式适合安装没有依赖的python包，如果有依赖使用conda install会报错，需要搭建一个内部仓库。
8.执行如下命令，激活python3运行环境:
[root@ip-172-31-21-45 ec2-user]# source activate py3
3.在CM配置Python环境变量
1.通过export设置python命令的安装路径：
export PYSPARK_PYTHON=/usr/local/anaconda3/bin/python
export PYSPARK_DRIVER_PYTHON=/usr/local/anaconda3/bin/python
修改完成后，回到CM主页根据提示重启相关服务。
4.pyspark命令测试
1.获取kerberos凭证
2.使用Pyspark命令测试
x = sc.parallelize(1,2,3)
y = x.flatMap(lambda x: (x, 100*x, x**2))
print(x.collect())
print(y.collect())
5.使用spark-submit提交一个Pyspark作业
这个demo主要使用spark-submit提交pyspark job，模拟从hdfs中读取数据，并转换成DateFrame，然后注册表并执行SQL条件查询，将查询结果输出到hdfs中。
1.将测试数据上传至hdfs目录/tmp/examples/
执行put命令上传文件，因为集群启用了Kerberos，所以也要使用kinit获取用户凭证信息
people.txt示例数据：
[ec2-user@ip-172-31-26-80 pysparktest]$ klist
Ticket cache: FILE:/tmp/krb5cc_1000
Default principal: fayson@CLOUDERA.COM
Valid starting Expires Service principal
12/17/2017 00:53:31 12/18/2017 00:53:31 krbtgt/CLOUDERA.COM@CLOUDERA.COM
renew until 12/24/2017 00:53:31
[ec2-user@ip-172-31-26-80 pysparktest]$ hadoop fs -mkdir /tmp/examples/
[ec2-user@ip-172-31-26-80 pysparktest]$ hadoop fs -put people.txt /tmp/examples
[ec2-user@ip-172-31-26-80 pysparktest]$ hadoop fs -cat /tmp/examples/people.txt
2.将pyspark程序上传至CDH集群其中一个节点上，该节点部署了Spark的Gateway角色和Python3
PySparkTest2HDFS.py在pysparktest目录中，内容如下：
# 初始化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.使用spark-submit命令向集群提交PySpark作业
root@ip-172-31-26-80 pysparktest# spark-submit PySparkTest2HDFS.py
4.作业执行成功
查看Yarn界面
通过以上信息，可以看到作业执行成功。
5.查看生成的文件，如下图：
因为生成的是parquet文件，它是二进制文件，无法直接使用命令查看，所以我们可以在pyspark上验证文件内容是否正确.
我们上面使用spark-submit提交的任务使用sql查询条件是13到19岁，可以看到在pyspark上查询的数据是在这个区间的数据
parquetFile = sqlContext.read.parquet("/tmp/examples/teenagers")
parquetFile.registerTempTable("parquetTable")
teenagers = sqlContext.sql("select* from parquetTable")
teenagers.show()
6.PySpark写数据到MySQL
1.将上面的作业增加如下代码
# 初始化sqlContext
from pyspark import SparkConf,SparkContext
from pyspark.sql import SQLContext, Row
conf=(SparkConf().setAppName('PySparkTest2MySQL'))
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")
url = "jdbc:mysql://ip-172-31-22-86.ap-southeast-1.compute.internal:3306/test"
table = "teenagers"
prop = {"user":"root","password":"123456"}
teenagers.write.jdbc(url, table, "append", prop)2.在命令行加载MySQL的驱动包到Spark环境变量，然后执行命令
[ec2-user@ip-172-31-26-80 pysparktest]$ export SPARK_CLASSPATH=$SPARK_CLASSPATH:/opt/cloudera/parcels/CDH/lib/spark/lib/mysql-connector-java-5.1.30.jar
[ec2-user@ip-172-31-26-80 pysparktest]$ spark-submit PySparkTest2Mysql.py 
执行成功
3.使用Yarn查看作业是否运行成功
4.验证MySQL表中是否有数据
注意：这里将数据写入MySQL时需要在环境变量中加载MySQL的JDBC驱动包，MySQL表可以不存在，pyspark在写数据时会自动创建该表。
为天地立心，为生民立命，为往圣继绝学，为万世开太平。
温馨提示：要看高清无码套图，请使用手机打开并单击图片放大查看
