python对hdfs/spark读写操作(hdfs/pyspark)

By | 2019年5月16日

  • 本项目是在服务器上,利用python对hdfs和spark,进行读写操作。以下内容,是基于服务器上,已经搭建好的hadoop/spark基础上,所以参考性有限,请见谅。搭建系统,不是很熟悉,单机spark请参照window+spark+python,或者公众号书圈搜Windows和PC机上搭建Spark+Python开发环境的详细步骤
  • 主要内容:利用python将数据保存到hadoop的hdfs目录下,利用spark读取hdfs目录下内容,然后做sql统计。

(一)、库安装

基于python3.5/python3.6

pip3 install hdfs
pip3 install pyspark

(二)、加载数据到hdfs

利用python,通过hdfs模块,连接集群,然后上传数据到hdfs目录。
关于python读写hdfs数据,请参照:hdfs数据读写

from hdfs.client import Client#hdfs模块
client = Client("http://192.177.6.17:50070",root="/",session=False)#服务器IP,端口,关闭session,避免一直开启
client.upload("/tmp/test1/","/home/test1/5min.csv")#第一个为hadoop中hdfs目录,第二个为文件目录

运行结果如下:

hadoop fs -ls /test1#显示目录下文件

在这里插入图片描述

(三)、sparksql读hdfs数据

通过pyspark模块,调用sparksql,读取hdfs目录数据,然后做统计。
参考:pyspark读取hdfs, python sparksql函数

from pyspark.sql import SQLContext
sc = SparkContext()
sqlcontext = SQLContext(sc)
#format后面为告诉程序读取csv格式,load后面为hdfs地址,hdfs后面跟着hadoop的名字,然后文件目录(这块有点懵,如果报错,跟着报错查修)
data = sqlcontext.read.format("com.databricks.spark.csv").\
    options(header="true",inferschema="true").\
    load("hdfs://cdsw1.richstone.com/test1/5min.csv")
data.show(5)
result = data.groupBy("type").count().orderBy("count")
result.show()

原始表结构:
在这里插入图片描述
查询结果为,注意运行最好用命令:spark-submit xxx.py
在这里插入图片描述

发表评论