Spark常用的4种读取和存储数据的格式 & 4种存储文件系统

2024-06-03 15:44| 来源: 网络整理| 查看: 265

Spark的数据读取与存储，常规来看可有分为两个维度，文件格式和文件系统：

文件格式：Text文件、Json文件、Sequence文件和Object文件

文件系统：linux本地文件系统、HDFS、HBase、MySQL数据库

一、文件格式类

1. Text文件

（1）读取Text文件 sc.textFile("path")

（2）创建RDD保存为Text文件

2. Json文件

3. Sequence文件（针对key-value类型RDD）

（1）创建RDD保存为Sequence文件

（2）读取Sequence文件 sc.sequenceFile[ ]("path")

4. Object文件

（1）将pairRDD保存为Object文件

（2）读取Object文件 sc.objectFile[ ]("path")

一、文件格式类 1. Text文件（1）读取Text文件 sc.textFile("path")

从HDFS的/目录下读取demo1.txt文件

scala> val hdfsFile = sc.textFile("hdfs://hadoop100:9000/demo1.txt") hdfsFile: org.apache.spark.rdd.RDD[String] = hdfs://hadoop100:9000/demo1.txt MapPartitionsRDD[1] at textFile at :24 scala> hdfsFile.collect res0: Array[String] = Array(diyihang) （2）创建RDD保存为Text文件 scala> hdfsFile.saveAsTextFile("hdfs://hadoop100:9000/demo/demo2.txt") scala> hdfsFile.saveAsTextFile("/opt/module/spark/demo2")

前者保存到HDFS中，后者保存到当前文件系统中

在50070端口，

在本地linux路径下，

[kevin@hadoop100 demo2]$ pwd /opt/module/spark/demo2 [kevin@hadoop100 demo2]$ ll 总用量 4 -rw-r--r-- 1 kevin kevin 9 2月 19 09:33 part-00000 -rw-r--r-- 1 kevin kevin 0 2月 19 09:33 part-00001 -rw-r--r-- 1 kevin kevin 0 2月 19 09:33 part-00002 -rw-r--r-- 1 kevin kevin 0 2月 19 09:33 _SUCCESS

可以看出都是分了3个区，与RDD的分区一致

scala> hdfsFile.partitions.length res11: Int = 3

一个分区对应一个task，一般而言，saveAsTextFile会按照执行task的多少生成多少个文件，比如part-00000一直到part-0000n，n自然就是task的个数，亦即是最后的stage的分区数。

2. Json文件

应用中多是采用SparkSQL处理Json文件，使用时需要导入解析json的包import scala.util.parsing.json.JSON，需要解析json文件

1. 首先写一个Json文件传输到HDFS中 {"name":"Michael","age":null} {"name":"Andy","age":"30"} {"name":"Justin","age":"20"} [kevin@hadoop100 datas]$ hadoop fs -put /opt/module/datas/people.json / 2. 导入解析json所需的包 scala> import scala.util.parsing.json.JSON 3. 读取Json文件 scala> val jsonFile = sc.textFile("hdfs://hadoop100:9000/people.json") jsonFile: org.apache.spark.rdd.RDD[String] = hdfs://hadoop100:9000/people.json MapPartitionsRDD[15] at textFile at :25 4. 解析Json文件 scala> val result = jsonFile.map(JSON.parseFull) result: org.apache.spark.rdd.RDD[Option[Any]] = MapPartitionsRDD[16] at map at :27 5. 打印 scala> result.collect res13: Array[Option[Any]] = Array(Some(Map(name -> Michael, age -> null)), Some(Map(name -> Andy, age -> 30)), Some(Map(name -> Justin, age -> 20)))

3. Sequence文件

Sequence序列化文件针对key-value类型的RDD

（1）创建RDD保存为Sequence文件 scala> val rdd = sc.makeRDD(Array((1,2),(3,4),(5,6))) rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[17] at makeRDD at :25 scala> rdd.saveAsSequenceFile("/opt/module/spark/seqFile")

保存的Sequence文件如下

[kevin@hadoop100 seqFile]$ ll 总用量 32 -rw-r--r-- 1 kevin kevin 92 2月 19 10:03 part-00000 -rw-r--r-- 1 kevin kevin 92 2月 19 10:03 part-00001 -rw-r--r-- 1 kevin kevin 108 2月 19 10:03 part-00002 -rw-r--r-- 1 kevin kevin 92 2月 19 10:03 part-00003 -rw-r--r-- 1 kevin kevin 92 2月 19 10:03 part-00004 -rw-r--r-- 1 kevin kevin 108 2月 19 10:03 part-00005 -rw-r--r-- 1 kevin kevin 92 2月 19 10:03 part-00006 -rw-r--r-- 1 kevin kevin 108 2月 19 10:03 part-00007 -rw-r--r-- 1 kevin kevin 0 2月 19 10:03 _SUCCESS （2）读取Sequence文件 sc.sequenceFile[ ]("path") scala> val seq = sc.sequenceFile[Int,Int]("/opt/module/spark/seqFile") seq: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[20] at sequenceFile at :25 scala> seq.collect res16: Array[(Int, Int)] = Array((5,6), (3,4), (1,2)) 4. Object文件

对象文件是将对象序列化后保存的文件，采用Java的序列化机制

（1）将pairRDD保存为Object文件 scala> val rdd = sc.makeRDD(Array(1,2,3,4,5)) rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[21] at makeRDD at :25 scala> rdd.saveAsObjectFile("/opt/module/spark/objectFile")

保存的文件如下

[kevin@hadoop100 objectFile]$ ll 总用量 32 -rw-r--r-- 1 kevin kevin 95 2月 19 10:13 part-00000 -rw-r--r-- 1 kevin kevin 138 2月 19 10:13 part-00001 -rw-r--r-- 1 kevin kevin 95 2月 19 10:13 part-00002 -rw-r--r-- 1 kevin kevin 138 2月 19 10:13 part-00003 -rw-r--r-- 1 kevin kevin 138 2月 19 10:13 part-00004 -rw-r--r-- 1 kevin kevin 95 2月 19 10:13 part-00005 -rw-r--r-- 1 kevin kevin 138 2月 19 10:13 part-00006 -rw-r--r-- 1 kevin kevin 138 2月 19 10:13 part-00007 -rw-r--r-- 1 kevin kevin 0 2月 19 10:13 _SUCCESS （2）读取Object文件 sc.objectFile[ ]("path") scala> val objFile = sc.objectFile[Int]("/opt/module/spark/objectFile") objFile: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[25] at objectFile at :25 scala> objFile.collect res19: Array[Int] = Array(3, 5, 1, 4, 2)

【本文地址】

公司简介

联系我们