PySpark读取数据与保存

Spark支持多种数据源如下

• 文件格式与文件系统
  对于存储在本地文件系统或分布式文件系统（HDFS、NFS）等的数据，Spark可以访问很多种不同的文件格式，包括文本文件、JSON、SequenceFile以及protocol buffer
• spark sql中结构化数据源
• 数据库与键值存储

文件格式

格式名称

结构化

备注

文本文件

否

普通的文本文件，每一行一条记录

JSON

半结构化

常见的基于文本的格式，半结构化

CSV

是

非常常见的基于文本的格式，通常的电子表格应用中使用

SequenceFiles

是

一种用于键值对数据常见Hadoop文件格式

Protocol buffers

是

一种快速，节约空间的跨语言格式

对象文件

是

用来将Spark作业中的数据存储下来以让共享的代码读取·

文件保存

from pyspark import SparkContext,SparkConf
if __name__ == '__main__':
conf=SparkConf().setAppName("save file").setMaster("local")
sc=SparkContext(conf=conf)
rdd=sc.parallelize(range(1,1000))
rdd.saveAsTextFile("new_rdd")

sql读取保存文件

from pyspark import SparkContext,SparkConf
from pyspark.sql import SparkSession
import sys
if __name__ == '__main__':
#配置spark环境
conf=SparkConf().setAppName("save file").setMaster("local")
sc=SparkContext(conf=conf)
rdd=sc.textFile('dict.txt',2).map(lambda x:x.strip('\n').split())
#配置DataFrame环境
spark=SparkSession.builder.appName("to json").getOrCreate()
df=spark.createDataFrame(rdd,['id','freq','label'])
df.write.csv('data_tocsv',mode='overwrite')
df.write.json('dict_tojson',mode='overwrite')
df.write.parquet('dict_toparquet',mode='overwrite')
df.write.saveAsTable("dict_totabel")

其SparkSession模块包含SQLContext和HiveContext来分别执行配置