Spark 数据读取和保存_val sparkconf = new sparkconf() .setmaster("local[

文本文件

val hdfsFile = sc.textFile("hdfs://hadoop01:9000/employee.txt")

hdfsFile.saveAsTextFile("/employeeOut")
1
2
3

JSON文件

每一行是一条JSON串

import scala.util.parsing.json.JSON

val json = sc.textFile("/employee.json")

val result  = json.map(JSON.parseFull)
1
2
3
4
5

Sequence文件

Sequence文件是Hadoop用来存储key-value二进制形式数据的文件

scala> val sequenceRdd = sc.parallelize(Array((1,2),(3,4),(5,6)))

scala> sequenceRdd.saveAsSequenceFile("file:///home/hadoop/spark/seqdata")

scala> val seq = sc.sequenceFile[Int,Int]("file:///opt/module/spark/seqdata")

scala> seq.collect
1
2
3
4
5
6
7

对象文件

数据会被序列化

scala> val rdd = sc.parallelize(Array(1,2,3))

scala> rdd.saveAsObjectFile("file:///home/hadoop/spark/objectdata")

scala> val obj = sc.objectFile[Int]("file:///home/hadoop/spark/objectdata")

scala> obj.collect
1
2
3
4
5
6
7

Hadoop API读写

旧版

scala> import org.apache.hadoop.io.{
   IntWritable, Text}

scala> import org.apache.hadoop.mapred.TextOutputFormat

scala> val content = sc.parallelize(Array(("laozhang",22),("laoli",18)))

scala> content.saveAsHadoopFile("hdfs://hadoop01:9000/test",classOf[Text],classOf[IntWritable],classOf[TextOutputFormat[Text,IntWritable]])
1
2
3
4
5
6
7
8

val conf = new SparkConf().setMaster(