首页
学习
活动
专区
工具
TVP
发布
精选内容/技术社群/优惠产品,尽在小程序
立即前往

如何在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-45ec2-user]# conda create --unknown --offline -n py3 python=3.5

注意:这里创建python3环境时使用了离线模式,即--offline参数,以及—unknown,这种方式适合安装没有依赖的python包,如果有依赖使用conda install会报错,需要搭建一个内部仓库。

8.执行如下命令,激活python3运行环境:

[root@ip-172-31-21-45ec2-user]# source activate py3

3.在CM配置Python环境变量

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

exportPYSPARK_PYTHON=/usr/local/anaconda3/bin/python

exportPYSPARK_DRIVER_PYTHON=/usr/local/anaconda3/bin/python

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

4.pyspark命令测试

1.获取kerberos凭证

2.使用Pyspark命令测试

x= sc.parallelize([1,2,3])

y= x.flatMap(lambdax:(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-80pysparktest]$klist

Ticketcache: FILE:/tmp/krb5cc_1000

Defaultprincipal: fayson@CLOUDERA.COM

Validstarting Expires Service principal

12/17/201700:53:31 12/18/2017 00:53:31 krbtgt/CLOUDERA.COM@CLOUDERA.COM

renewuntil12/24/2017 00:53:31

[ec2-user@ip-172-31-26-80pysparktest]$hadoop fs -mkdir /tmp/examples/

[ec2-user@ip-172-31-26-80pysparktest]$hadoop fs -put people.txt /tmp/examples

[ec2-user@ip-172-31-26-80pysparktest]$hadoop fs -cat /tmp/examples/people.txt

2.将pyspark程序上传至CDH集群其中一个节点上,该节点部署了Spark的Gateway角色和Python3

PySparkTest2HDFS.py在pysparktest目录中,内容如下:

#初始化sqlContext

frompysparkimportSparkConf,SparkContext

frompyspark.sqlimportSQLContext, Row

conf=(SparkConf().setAppName('PySparkTest2HDFS'))

sc=SparkContext(conf=conf)

sqlContext = SQLContext(sc)

#加载文本文件并转换成Row.

lines = sc.textFile("/tmp/examples/people.txt")

parts = lines.map(lambdal: l.split(","))

people = parts.map(lambdap: Row(name=p[],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

#将查询结果保存至hdfs中

teenagers.write.save("/tmp/examples/teenagers")

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

[root@ip-172-31-26-80pysparktest]# 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

frompysparkimportSparkConf,SparkContext

frompyspark.sqlimportSQLContext, Row

conf=(SparkConf().setAppName('PySparkTest2MySQL'))

sc=SparkContext(conf=conf)

sqlContext = SQLContext(sc)

#加载文本文件并转换成Row.

lines = sc.textFile("/tmp/examples/people.txt")

parts = lines.map(lambdal: l.split(","))

people = parts.map(lambdap: Row(name=p[],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

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-80pysparktest]$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-80pysparktest]$spark-submit PySparkTest2Mysql.py

执行成功

3.使用Yarn查看作业是否运行成功

4.验证MySQL表中是否有数据

注意:这里将数据写入MySQL时需要在环境变量中加载MySQL的JDBC驱动包,MySQL表可以不存在,pyspark在写数据时会自动创建该表。

为天地立心,为生民立命,为往圣继绝学,为万世开太平。

温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看。

您可能还想看

安装

安全

数据科学

其他

  • 发表于:
  • 原文链接http://kuaibao.qq.com/s/20171217G0L34100?refer=cp_1026
  • 腾讯「腾讯云开发者社区」是腾讯内容开放平台帐号(企鹅号)传播渠道之一,根据《腾讯内容开放平台服务协议》转载发布内容。
  • 如有侵权,请联系 cloudcommunity@tencent.com 删除。

扫码

添加站长 进交流群

领取专属 10元无门槛券

私享最新 技术干货

扫码加入开发者社群
领券