温馨提示：如果使用电脑查看图片不清晰，可以使用手机打开文章单击文中的图片放大查看高清原图。
Fayson的github： https://github.com/fayson/cdhproject
提示：代码块部分可以左右滑动查看噢
1
文档编写目的
当前有很多工具辅助大数据分析，但最受环境的就是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
2
基于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。
代码语言：javascript复制[root@ip-172-31-13-38 ~]$ sudo yum -y install bzip2
5.安装完后，提示设置anaconda的PATH路径，这里需要设置全局路径，因为要确保pyspark任务提交过来之后可以使用python3，所以输入“no”，重新设置PATH
6.设置全局的anaconda3的PATH
代码语言：javascript复制[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版本
3
在CM配置Spark2的Python环境
1.通过export设置python命令的安装路径：
代码语言：javascript复制export PYSPARK_PYTHON=/opt/cloudera/anaconda3/bin/python
export PYSPARK_DRIVER_PYTHON=/opt/cloudera/anaconda3/bin/python
修改完成后，回到CM主页根据提示重启相关服务。
4
pyspark命令测试
1.获取kerberos凭证
2.使用Pyspark2命令测试
代码语言：javascript复制x = sc.parallelize([1,2,3])
y = x.flatMap(lambda x: (x, 100*x, x**2))
print(x.collect())
print(y.collect())
5
提交一个Pyspark作业
这个demo主要使用spark2-submit提交pyspark job，模拟从hdfs中读取数据，并转换成DateFrame，然后注册为临时表并执行SQL条件查询，将查询结果输出到hdfs中。
1.将测试数据上传至hdfs目录/tmp/examples/
代码语言：javascript复制[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程序内容如下：
代码语言：javascript复制# 初始化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作业
代码语言：javascript复制[root@ip-172-31-13-38 ~]# spark2-submit PySparkTest2HDFS.py
4.作业执行成功
查看Spark2的HistoryServer
通过以上信息，可以看到作业执行成功。
5. 查看生成的文件，如下图：
因为生成的是parquet文件，它是二进制文件，无法直接使用命令查看，所以我们可以在pyspark上验证文件内容是否正确.
我们上面使用spark2-submit提交的任务使用sql查询条件是3到4岁，可以看到在pyspark2上查询的数据是在这个区间的数据
代码语言：javascript复制parquetFile = sqlContext.read.parquet("/tmp/examples/teenagers")
parquetFile.registerTempTable("parquetTable")
teenagers = sqlContext.sql("select * from parquetTable")
teenagers.show()
提示：代码块部分可以左右滑动查看噢
为天地立心，为生民立命，为往圣继绝学，为万世开太平。 温馨提示：如果使用电脑查看图片不清晰，可以使用手机打开文章单击文中的图片放大查看高清原图。
推荐关注Hadoop实操，第一时间，分享更多Hadoop干货，欢迎转发和分享。
