Spark——数据读取与保存

Spark的数据读取与保存

- 一、文件类数据读取与保存
- - 1.1 Text 文件
  - 1.2 Json 文件
  - 1.3 Sequence 文件
  - 1.4 对象文件
- 二、文件系统类数据读取与保存
- - 2.1 HDFS
  - 2.2 MySQL 数据库连接
  - 2.3 HBase 数据库

Spark 的数据读取及数据保存可以从两个维度来作区分：文件格式以及文件系统。

文件格式分为：Text 文件、Json 文件、Csv 文件、Sequence 文件以及 Object 文件；
文件系统分为：本地文件系统、HDFS、HBASE 以及数据库。

一、文件类数据读取与保存

1.1 Text 文件

(1）数据读取:textFile(String)

scala> val hdfsFile = sc.textFile("hdfs://hadoop102:9000/fruit.txt")
hdfsFile: org.apache.spark.rdd.RDD[String] = hdfs://hadoop102:9000/fruit.txt
MapPartitionsRDD[21] at textFile at :24

(2）数据保存: saveAsTextFile(String)

scala> hdfsFile.saveAsTextFile("/fruitOut")

在这里插入图片描述

1.2 Json 文件

如果 JSON 文件中每一行就是一个 JSON 记录，那么可以通过将 JSON 文件当做文本文件来读取，然后利用相关的 JSON 库对每一条数据进行 JSON 解析。

注意：使用 RDD 读取 JSON 文件处理很复杂，同时 SparkSQL 集成了很好的处理JSON 文件的方式，所以应用中多是采用SparkSQL 处理 JSON 文件。

（1）导入解析 json 所需的包

scala> import scala.util.parsing.json.JSON

（2）上传 json 文件到 HDFS

[atguigu@hadoop102 spark]$ hadoop fs -put ./examples/src/main/resources/people.json /

（3）读取文件

scala> val json = sc.textFile("/people.json")
json: org.apache.spark.rdd.RDD[String] = /people.json MapPartitionsRDD[8] at textFile at:24

（4）解析 json 数据

scala> val result = json.map(JSON.parseFull)
result: org.apache.spark.rdd.RDD[Option[Any]] = MapPartitionsRDD[10] at map at :27

（5）打印

scala> result.collect
res11: Array[Option[Any]] = Array(Some(Map(name -> Michael)), Some(Map(name -> Andy, age
-> 30.0)), Some(Map(name -> Justin, age -> 19.0)))

1.3 Sequence 文件

SequenceFile 文件是 Hadoop 用来存储二进制形式的 key-value 对而设计的一种平面文件(Flat File)。Spark 有专门用来读取 SequenceFile 的接口。在 SparkContext 中，可以调用 sequenceFile[ keyClass, valueClass](path)。

注意：SequenceFile 文件只针对 PairRDD

（1）创建一个 RDD

scala> val rdd = sc.parallelize(Array((1,2),(3,4),(5,6)))
rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[13] at parallelize at:24

（2）将 RDD 保存为 Sequence 文件

scala> rdd.saveAsSequenceFile("file:///opt/module/spark/seqFile")

（3）查看该文件

[atguigu@hadoop102 seqFile]$ pwd
/opt/module/spark/seqFile
[atguigu@hadoop102 seqFile]$ ll
总用量 8
-rw-r--r-- 1 atguigu atguigu 108 10 月 9 10:29 part-00000
-rw-r--r-- 1 atguigu atguigu 124 10 月 9 10:29 part-00001
-rw-r--r-- 1 atguigu atguigu 0 10 月 9 10:29 _SUCCESS
[atguigu@hadoop102 seqFile]$ cat part-00000
SEQ org.apache.hadoop.io.IntWritable org.apache.hadoop.io.IntWritableط

（4）读取 Sequence 文件

scala> val seq = sc.sequenceFile[Int,Int]("file:///opt/module/spark/seqFile")
seq: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[18] at sequenceFile at:24

（5）打印读取后的 Sequence 文件

scala> seq.collect
res14: Array[(Int, Int)] = Array((1,2), (3,4), (5,6))

1.4 对象文件

对象文件是将对象序列化后保存的文件，采用 Java 的序列化机制。可以通过objectFile[k,v](path) 函数接收一个路径，读取对象文件，返回对应的 RDD，也可以通过调用 saveAsObjectFile() 实现对对象文件的输出。因为是序列化所以要指定类型。

（1）创建一个 RDD

scala> val rdd = sc.parallelize(Array(1,2,3,4))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[19] at parallelize at :24

（2）将 RDD 保存为 Object 文件

scala> rdd.saveAsObjectFile("file:///opt/module/spark/objectFile")

（3）查看该文件

[atguigu@hadoop102 objectFile]$ pwd
/opt/module/spark/objectFile
[atguigu@hadoop102 objectFile]$ ll
总用量 8
-rw-r--r-- 1 atguigu atguigu 142 10 月 9 10:37 part-00000
-rw-r--r-- 1 atguigu atguigu 142 10 月 9 10:37 part-00001
-rw-r--r-- 1 atguigu atguigu 0 10 月 9 10:37 _SUCCESS
[atguigu@hadoop102 objectFile]$ cat part-00000
SEQ!org.apache.hadoop.io.NullWritable"org.apache.hadoop.io.BytesWritableW@`l

（4）读取 Object 文件

scala> val objFile = sc.objectFile[(Int)]("file:///opt/module/spark/objectFile")
objFile: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[31] at objectFile at :24

（5）打印读取后的 Sequence 文件

scala> objFile.collect
res19: Array[Int] = Array(1, 2, 3, 4)

二、文件系统类数据读取与保存

2.1 HDFS

Spark 的整个生态系统与 Hadoop 是完全兼容的，所以对于 Hadoop 所支持的文件类型或者数据库类型，Spark 也同样支持。另外,由于 Hadoop 的 API 有新旧两个版本，所以 Spark 为了能够兼容 Hadoop 所有的版本，也提供了两套创建操作接口。对于外部存储创建操作而言，hadoopRDD 和 newHadoopRDD 是最为抽象的两个函数接口，主要包含以下四个参数：

（1）输入格式(InputFormat): 制定数据输入的类型,如 TextInputFormat 等,新旧两个版本所引用的版本分别是 org.apache.hadoop.mapred.InputFormat 和org.apache.hadoop.mapreduce.InputFormat(NewInputFormat)

（2）键类型: 指定[K,V]键值对中 K 的类型

（3）值类型: 指定[K,V]键值对中 V 的类型

（4）分区值: 指定由外部存储生成的 RDD 的 partition 数量的最小值,如果没有指定,系统会使用默认值 defaultMinSplits

注意：其他创建操作的 API 接口都是为了方便最终的 Spark 程序开发者而设置的，是这两个接口的高效实现版本。例如：对于textFile 而言，只有 path 这个指定文件路径的参数，其他参数在系统内部指定了默认值。

1、在 Hadoop 中以压缩形式存储的数据，不需要指定解压方式就能够进行读取，因为Hadoop 本身有一个解压器会根据压缩文件的后缀推断解压算法进行解压。

2、如果用 Spark 从 Hadoop 中读取某种类型的数据不知道怎么读取的时候，上网查找一个使用 map-reduce 的时候是怎么读取这种这种数据的，然后再将对应的读取方式改写成上面的hadoopRDD 和 newAPIHadoopRDD 两个类就行了。

2.2 MySQL 数据库连接

支持通过 Java JDBC 访问关系型数据库，需要通过 JdbcRDD 进行，示例如下:

（1）添加依赖

<dependency><groupId> mysql</groupId><artifactId>mysql-connector-java </artifactId><version>5.1.27</version>
</dependency>

（2）Mysql 读取：

import java.sql.DriverManager
import org.apache.spark.rdd.JdbcRDD
import org.apache.spark.{SparkConf, SparkContext}
object MysqlRDD {
def main(args: Array[String]): Unit = {//1.创建 spark 配置信息val sparkConf: SparkConf = new
SparkConf().setMaster("local[*]").setAppName("JdbcRDD")//2.创建 SparkContextval sc = new SparkContext(sparkConf)//3.定义连接 mysql 的参数val driver = "com.mysql.jdbc.Driver"val url = "jdbc:mysql://hadoop102:3306/rdd"val userName = "root"val passWd = "000000"//创建 JdbcRDDval rdd = new JdbcRDD(sc, () => {Class.forName(driver)DriverManager.getConnection(url, userName, passWd)},"select * from `rddtable` where `id`>=?;",1,10,1,r => (r.getInt(1), r.getString(2)))//打印最后结果println(rdd.count())rdd.foreach(println)sc.stop()
}
}

Mysql 写入：

def main(args: Array[String]) {val sparkConf = new SparkConf().setMaster("local[2]").setAppName("HBaseApp")val sc = new SparkContext(sparkConf)val data = sc.parallelize(List("Female", "Male","Female"))data.foreachPartition(insertData)
}
def insertData(iterator: Iterator[String]): Unit = {
Class.forName ("com.mysql.jdbc.Driver").newInstance()val conn = java.sql.DriverManager.getConnection("jdbc:mysql://master01:3306/rdd", "root",
"hive")iterator.foreach(data => {val ps = conn.prepareStatement("insert into rddtable(name) values (?)")ps.setString(1, data)ps.executeUpdate()})
}

2.3 HBase 数据库

由于 org.apache.hadoop.hbase.mapreduce.TableInputFormat 类的实现，Spark 可以通过Hadoop 输入格式访问 HBase。这个输入格式会返回键值对数据，其中键的类型为 org.apache.hadoop.hbase.io.ImmutableBytesWritable，而值的类型为org.apache.hadoop.hbase.client.Result。