spark支持多种数据源，从总体来分分为两大部分：文件系统和数据库。
文件系统主要有本地文件系统、Amazon S3、HDFS等。
文件系统中存储的文件有多种存储格式。spark支持的一些常见格式有：
| 格式名称 | 结构化 | 说明 |
|---|---|---|
| 文件文件 | 否 | 普通文件文件，每行一条记录 |
| JSON | 半结构化 | 常见的基于文本的半结构化数据 |
| CSV | 是 | 常见的基于文本的格式，在电子表格应用中使用 |
| SequenceFiles | 是 | 一种用于键值对数据的常见Hadoop文件格式 |
读取
读取单个文件，参数为文件全路径，输入的每一行都会成为RDD的一个元素。
input = sc.textFile("file://opt/module/spark/README.md")val input = sc.textFile("file://opt/module/spark/README.md")JavaRDD<String> input = sc.textFile("file://opt/module/spark/README.md")val input = sc.wholeTextFiles("file://opt/module/spark/datas")
val result = input.mapValues{
y => {
val nums = y.split(" ").map(x => x.toDouble)
nums.sum / nums.size.toDouble
}
}输出文本文件时，可使用saveAsTextFile()方法接收一个目录，将RDD中的内容输出到目录中的多个文件中。
```
result.saveAsTextFile(outputFile)
```读取
import json
...
input = sc.textFile("file.json")
data = input.map(lambda x: json.loads(x))import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
...
case class Person(name: String, lovesPandas: Boolean)
...
val input = sc.textFile("file.json")
val mapper = new ObjectMapper()
mapper.registerModule(DefaultScalaModule)
val result = input.flatMap(record => {
try {
Some(mapper.readValue(record, classOf[Person]))
} catch {
case e: Exception => None
}
})class ParseJson implements FlatMapFunction<Iterator<String>, Person> {
public Iterable<Person> call(Iterator<String> lines) throws Exception {
ArrayList<Person> people = new ArrayList<Person>();
ObjectMapper mapper = new ObjectMapper();
while(lines.hasNext()) {
String line = lines.next();
try {
people.add(mapper.readValue(line, Person.class));
} catch(Exception e) {
//跳过失败的数据
}
}
return people;
}
}
JavaRDD<String> input = sc.textFile("file.json");
JavaRDD<Person> result = input.mapPartitions(new ParseJson());写入
(data.filter(lambda x: x["lovesPandas"]).map(lambda x: json.dumps(x)).saveAsTextFile(outputFile)result.filter(p => p.lovesPandas).map(mapper.writeValueAsString(_)).saveAsTextFile(outputFile)class WriteJson implements FlatMapFunction<Iterator<Person>, String> {
public Iterable<String> call(Iterator<Person> people) throws Exception {
ArrayList<String> text = new ArrayList<String>();
ObjectMapper mapper = new ObjectMapper();
while(people.hasNext()) {
Person person = people.next();
text.add(mapper.writeValueAsString(person));
}
return text;
}
}
JavaRDD<Person> result = input.mapPartitions(new ParseJson()).filter(new LikesPandas());
JavaRDD<String> formatted = result.mapPartitions(new WriteJson());
formatted.saveAsTextFile(outfile);CSV与TSV文件每行都有固定的字段，字段之间使用分隔符（CSV使用逗号；tsv使用制表符）分隔。
读取
将csv或tsv文件当作普通文本文件读取，然后使用响应的解析器进行解析，同json处理方式。
python使用内置库读取csv
import csv
import StringIO
...
def loadRecord(line):
input = StringIO.StringIO(line)
reader = csv.DictReader(input, fieldnames=["name","favouriteAnimal"])
return reader.next()
"""读取每行记录"""
input = sc.textFile(inputFile).map(loadRecord)def loadRecords(fileNameContents):
input = StringIO.StringIO(fileNameContents[1])
reader = csv.DictReader(input, fieldnames=["name","favoriteAnimal"])
return reader
"""读取整个文件"""
fullFileData = sc.wholeTextFiles(inputFile).flatMap(loadRecords)scala使用opencsv库读取csv
import Java.io.StringReader
import au.com.bytecode.opencsv.CSVReader
...
val input = sc.textFile(inputFile)
val result = input.map{
line => {
val reader = new CSVReader(new StringReader(line))
reader.readNext()
}
}case class Person(name: String, favoriteAnimal: String)
val input = sc.wholeTextFiles(inputFile)
val result = input.flatMap(
case(_, txt) => {
val reader = new CSVReader(new StringReader(txt))
reader.readAll().map(x => Person(x(0), x(1)))
}java使用opencsv库读取csv
import Java.io.StringReader
import au.com.bytecode.opencsv.CSVReader
...
public static class ParseLine implements Function<String, String[]> {
public String[] call(String line) throws Exception {
CSVReader reader = new CSVReader(new StringReader(line));
return reader.readNext();
}
}
JavaPairRDD<String[]> csvData = sc.textFile(inputFile).map(new ParseLine());public static class ParseLine implements FlatMapFunction<Tuple2<String, String>, String[]> {
public Iterable<String[]> call(Tuple2<String, String> file) throws Exception {
CSVReader reader = new CSVReader(new StringReader(file._2);
return reader.readAll();
}
}
JavaRDD<String[]> keyedRDD = sc.wholeTextFiles(inputFile).flatMap(new ParseLine());写入
def writeRecords(records):
output = StringIO.StringIO()
writer = csv.DictWriter(output, fieldnames=["name", "favoriteAnimal"])
for record in records:
writer.writerow(record)
return [output.getValue()]
pandaLovers.mapPartitions(writeRecords).saveAsTextFile(outputFile)pandasLovers.map(person => List(person.name, person.favoriteAnimal).toArray).mapPartitions{
people => {
val stringWriter = new StringWriter()
val csvWriter = new CSVWriter(stringWriter)
csvWriter.writeAll(people.toList)
Iterator(stringWriter.toString)
}
}.saveAsTextFile(outFile)SequenceFile是键值对形式的常用Hadoop数据格式。由于Hadoop使用一套自定义的序列化框架，因此SequenceFile的键值对类型需实现Hadoop的Writable接口。
读取
data = sc.sequenceFile(inFile, "org.apache.hadoop.io.Text", "org.apache.hadoop.io.IntWritable")val data = sc.sequenceFile(inFile, classOf[Text], classOf[IntWritable]).map{case (x, y) => (x.toString, y.get())} public static class ConvertToNativeTypes implements PairFunction<Tuple2<Text, IntWritable>, String, Integer> {
public Tuple2<String, Integer> call(Tuple2<Text, IntWritable> record) {
return new Tuple2(record._1.toString(), record._2.get());
}
}
JavaPairRDD<String, Integer> result = sc.sequenceFile(fileName, Text.class, IntWritable.class).mapToPair(new ConvertToNativeTypes());写入
data = sc.parallelize([("Panda", 3), ("Kay", 6), ("Snail", 2)])
data.saveAsSequeceFile(outputFile)val data = sc.parallelize(List(("Panda", 3), ("Kay", 6), ("Snail", 2)))
data.saveAsSequenceFile(outputFile)public static class ConvertToWritableTypes implements PairFunction<Tuple2<String, Integer>, Text, IntWritable> {
public Tuple2<Text, IntWritable> call(Tuple2<String, Integer> record) {
return new Tuple2(new Text(record._1), new IntWritable(record._2));
}
}
JavaPairRDD<Text, IntWritable> result = sc.parallelizePairs(input).mapToPair(new ConvertToNativeTypes());
result.saveAsHadoopFile(fileName, Text.class, IntWritable.class, SequenceFileOutputFormat.class);数据库主要分为关系型数据库（MySQL、PostgreSQL等）和非关系型数据库（HBase、ElasticSearch等）。
spark使用JDBC访问关系型数据库（MySQL、PostgreSQL等），只需要构建一个org.apache.spark.rdd.JdbcRDD即可。
def createConnection() = {
Class.forName("com.mysql.jdbc.Driver").newInstance()
DriverManager.getConnection("jdbc:mysql://localhost/test", "root", "root")
}
def extractValues(r: ResultSet) = {
(r.getInt(1), r.getString(2))
}
val data = new JdbcRDD(sc, createConnection,
"SELECT * FROM panda WHERE id >= ? AND id <= ?"),
lowerBound = 1, upperBound = 3,
numPartitions = 2, mapRow = extractValues)
println(data.collect().toList)spark通过Hadoop输入格式（org.apache.hadoop.hbase.mapreduce.TableInputFormat）访问HBase。这种格式返回键值对数据，键类型为org.apache.hadoop.hbase.io.ImmutableBytesWritable，值类型为org.apache.hadoop.hbase.client.Result。
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
val conf = HBaseConfiguration.create()
conf.set(TableInputFormat.INPUT_TABLE, "tablename")
val rdd = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], ClassOf[Result])spark使用ElasticSearch-Hadoop连接器从ElasticSearch中读写数据。ElasticSearch连接器依赖于SparkContext设置的配置项。ElasticSearch连接器也没有用到Spark封装的类型，而使用saveAsHadoopDataSet。
def mapWritableToInput(in: MapWritable): Map[String, String] = {
in.map{case (k, v) => (k.toString, v.toString)}.toMap
}
val jobConf = new JobConf(sc.hadoopConfiguration)
jobConf.set(ConfigurationOptions.ES_RESOURCE_READ, args[1])
jobConf.set(ConfigurationOptions.ES_NODES, args[2])
val currentTweets = sc.hadoopRDD(jobConf, classOf[EsInputFormat[Object, MapWritable]], classOf[Object], ClassOf[MapWritable])
val tweets = currentTweets.map{ case (key, value) => mapWritableToInput(value) }val jobConf = new JobConf(sc.hadoopConfiguration)
jobConf.set("mapred.output.format.class", "org.elasticsearch.hadoop.mr.EsOutFormat")
jobConf.setOutputCommitter(classOf[FileOutputCommitter])
jobConf.set(ConfigurationOptions.ES_RESOURCE_WRITE, "twitter/tweets")
jobConf.set(ConfigurationOptions.ES_NODES, "localhost")
FileOutputFormat.setOutputPath(jobConf, new Path("-"))
output.saveAsHadoopDataset(jobConf)忠于技术，热爱分享。欢迎关注公众号：java大数据编程，了解更多技术内容。
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
