3. Spark常见数据源
作者:互联网
*以下内容由《Spark快速大数据分析》整理所得。
读书笔记的第三部分是讲的是Spark有哪些常见数据源?怎么读取它们的数据并保存。
Spark有三类常见的数据源:
- 文件格式与文件系统:它们是存储在本地文件系统或分布式文件系统(比如 NFS、HDFS、Amazon S3 等)中的 数据,例如:文本文件、JSON、SequenceFile, 以及 protocol buffer。
- Spark SQL中的结构化数据源:它针对包括JSON和Apache Hive在内的结构化数据源。
- 数据库与键值存储:Spark 自带的库和一些第三方库,它们可以用来连接Cassandra、HBase、Elasticsearch以及JDBC源。
一、文件格式与文件系统
1. 文本文件
2. JSON
3. CSV
4. SequenceFile
二、Spark SQL中的结构化数据源
三、数据库与键值存储
一、文件格式与文件系统
1. 文本文件
文本文件读取:
# 方法1:文本文件读取 input = sc.textFile("file://home/holden/repos/sparks/README.md") # 方法2:如果文件足够小,同时读取整个文件,从而返回一个pair RDD,其中键时输入文件的文件名 input = sc.wholeTextFiles("file://home/holden/salesFiles")
文本文件保存:
result.saveAsTextFile(outputFile)
2. JSON
JSON读取:
# JSON读取 import json data = input.map(lambda x: json.loads(x))
JSON保存:
# JSON保存 - 举例选出喜爱熊猫的人 (data.filter(lambda x: x["lovesPandas"]).map(lambda x: json.dumps(x)).saveAsTextFile(outputFile)) # 保存文本文件 result.SaveAsTextFile(outputFilePath)
3. CSV
CSV读取:
import csv import StringIO
# CSV读取 - 如果数据字段均没有包括换行符,只能一行行读取 def loadRecord(line): """解析一行CSV记录""" input = StringIO.StringIO(line) reader = csv.DictReader(input, fieldnames=["name", "favouriteAnimal"]) return reader.next()
input = sc.textFile(inputFile).map(loadRecord)
# CSV读取 - 如果数据字段嵌有换行符,需要完整读入每个文件 def loadRecords(fileNameContents): """读取给定文件中的所有记录""" input = StringIO.StringIO(fileNameContents[1]) reader = csv.DictReader(input, fieldnames=["name", "favoriteAnimal"]) return reader
fullFileData = sc.wholeTextFiles(inputFile).flatMap(loadRecords)
CSV保存:
# CSV保存 def writeRecords(records): """写出一些CSV记录""" output = StringIO.StringIO() writer = csv.DictWriter(output, fieldnames=["names", "favoriteAnimal"]) for record in records: writer.writerow(record) return [output.getvalue()] pandaLovers.mapPartitions(writeRecords).saveAsTextFile(outputFile)
4. SequenceFile
SequenceFile读取:
# sc.sequenceFile(path, keyClass, valueClass)
data = sc.sequenceFile(inFile, "org.apache.hadoop.io.Text", "org.apache.hadoop.io.IntWritable")
SequenceFile保存(用Scala):
val data = sc.parallelize(List(("Pandas", 3), ("Kay", 6), ("Snail", 2))) data.saveAsSequenceFile(outputFile)
二、Spark SQL中的结构化数据源
三、数据库与键值存储
关于Cassandra、HBase、Elasticsearch以及JDBC源的数据库连接,详情请参考书本81-86页内容。
标签:读取,StringIO,数据源,常见,input,Spark,CSV 来源: https://www.cnblogs.com/alvinai/p/13710788.html