spark保存计算结果

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

文件格式分为&＃xff1a;Text 文件、Json 文件、Csv 文件、Sequence 文件以及 Object 文件&＃xff1b;

文件系统分为&＃xff1a;本地文件系统、HDFS、HBASE 以及数据库。

1&＃xff09;数据读取:textFile(String) 

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

2&＃xff09;数据保存: saveAsTextFile(String)

scala> hdfsFile.saveAsTextFile("/fruitOut")

4.1.2 Json 文件 

　　如果 JSON 文件中每一行就是一个 JSON 记录&＃xff0c;那么可以通过将 JSON 文件当做文本

文件来读取&＃xff0c;然后利用相关的 JSON 库对每一条数据进行 JSON 解析。

　　注意&＃xff1a;使用 RDD 读取 JSON 文件处理很复杂&＃xff0c;同时 SparkSQL 集成了很好的处理

JSON 文件的方式&＃xff0c;所以应用中多是采用 SparkSQL 处理 JSON 文件。 

&＃xff08;1&＃xff09;导入解析 json 所需的包 

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

&＃xff08;2&＃xff09;上传 json 文件到 HDFS

[lxl&＃64;hadoop102 spark]$ hadoop fs -put ./examples/src/main/resources/people.json /

&＃xff08;3&＃xff09;读取文件

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

&＃xff08;4&＃xff09;解析 json 数据

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

&＃xff08;5&＃xff09;打印

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

4.1.3 Sequence 文件 

　　SequenceFile 文件是 Hadoop 用来存储二进制形式的 key-value 对而设计的一种平面

文件(Flat File)。Spark 有专门用来读取 SequenceFile 的接口。在 SparkContext 中&＃xff0c;可以

调用 sequenceFile[ keyClass, valueClass](path)。

注意&＃xff1a;SequenceFile 文件只针对 PairRDD 

&＃xff08;1&＃xff09;创建一个 RDD

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

&＃xff08;2&＃xff09;将 RDD 保存为 Sequence 文件

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

&＃xff08;3&＃xff09;查看该文件 

[lxl&＃64;hadoop102 seqFile]$ pwd
/opt/module/spark/seqFile[lxl&＃64;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[lxl&＃64;hadoop102 seqFile]$ cat part-00000
SEQ org.apache.hadoop.io.IntWritable org.apache.hadoop.io.IntWritableط

&＃xff08;4&＃xff09;读取 Sequence 文件 

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

&＃xff08;5&＃xff09;打印读取后的 Sequence 文件

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

4.1.4 对象文件 &＃xff08;objectFile&＃xff09;

　　对象文件是将对象序列化后保存的文件&＃xff0c;采用 Java 的序列化机制。可以通过

objectFile[k,v](path) 函数接收一个路径&＃xff0c;读取对象文件&＃xff0c;返回对应的 RDD&＃xff0c;也可以通过调

用 saveAsObjectFile() 实现对对象文件的输出。因为是序列化所以要指定类型。 

&＃xff08;1&＃xff09;创建一个 RDD

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

&＃xff08;2&＃xff09;将 RDD 保存为 Object 文件 

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

&＃xff08;3&＃xff09;查看该文件

[lxl&＃64;hadoop102 object]$ pwd
/opt/module/spark/object

  [lxl&＃64;hadoop102 object]$ ll
  总用量 16
  -rw-r--r-- 1 lxl lxl 138 7月 8 03:12 part-00000
  -rw-r--r-- 1 lxl lxl 138 7月 8 03:12 part-00001
  -rw-r--r-- 1 lxl lxl 138 7月 8 03:12 part-00002
  -rw-r--r-- 1 lxl lxl 142 7月 8 03:12 part-00003
  -rw-r--r-- 1 lxl lxl 0 7月 8 03:12 _SUCCESS

  [lxl&＃64;hadoop102 object]$ cat part-00000
  SEQ!org.apache.hadoop.io.NullWritable"org.apache.hadoop.io.BytesWritabley.)a¬촲[IMº&＃96;&v겥xp

&＃xff08;4&＃xff09;读取 Object 文件 

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

&＃xff08;5&＃xff09;打印读取后的 Sequence 文件 

scala> objFile.collect
res19: Array[Int] &＃61; Array(1, 2, 3, 4)

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

4.2.1 HDFS 

　　Spark 的整个生态系统与 Hadoop 是完全兼容的,所以对于 Hadoop 所支持的文件类型

或者数据库类型,Spark 也同样支持.另外,由于 Hadoop 的 API 有新旧两个版本,所以 Spark 为

了能够兼容 Hadoop 所有的版本,也提供了两套创建操作接口.对于外部存储创建操作而

言,hadoopRDD 和 newHadoopRDD 是最为抽象的两个函数接口,主要包含以下四个参数.

　　1&＃xff09;输入格式(InputFormat): 制定数据输入的类型,如 TextInputFormat 等,新旧两个版本

所引用的版本分别是 org.apache.hadoop.mapred.InputFormat 和

org.apache.hadoop.mapreduce.InputFormat(NewInputFormat)

　　2&＃xff09;键类型: 指定[K,V]键值对中 K 的类型

　　3&＃xff09;值类型: 指定[K,V]键值对中 V 的类型

　　4&＃xff09;分区值: 指定由外部存储生成的 RDD 的 partition 数量的最小值,如果没有指定,系

统会使用默认值 defaultMinSplits

注意:其他创建操作的 API 接口都是为了方便最终的 Spark 程序开发者而设置的,是这两个

接口的高效实现版本.例如,对于 textFile 而言,只有 path 这个指定文件路径的参数,其他参数

在系统内部指定了默认值。

　　1.在 Hadoop 中以压缩形式存储的数据,不需要指定解压方式就能够进行读取,因为

Hadoop 本身有一个解压器会根据压缩文件的后缀推断解压算法进行解压.

　　2.如果用 Spark 从 Hadoop 中读取某种类型的数据不知道怎么读取的时候,上网查找一个

使用 map-reduce 的时候是怎么读取这种这种数据的,然后再将对应的读取方式改写成上面的

hadoopRDD 和 newAPIHadoopRDD 两个类就行了 

4.2.2 MySQL 数据库连接 

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

&＃xff08;1&＃xff09;添加依赖 

mysqlmysql-connector-java5.1.27

&＃xff08;2&＃xff09;Mysql 读取&＃xff1a; 

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

Mysql 写入&＃xff1a; 

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

spark-shell 中使用 JDBC 连接 Mysql&＃xff1a;

[lxl&＃64;hadoop102 spark]$ cp /opt/module/hive/lib/mysql-connector-java-5.1.27-bin.jar ./jars/

scala> val rdd &＃61; new org.apache.spark.rdd.JdbcRDD(sc, () &＃61;> {| Class.forName("com.mysql.jdbc.Driver")| java.sql.DriverManager.getConnection("jdbc:mysql://hadoop102:3306/rdd", "root", "000000") | },| "select * from &＃96;rddtable&＃96; where id >&＃61; ? and id <&＃61; ?;",| 1,| 10,| 1,| r &＃61;> (r.getInt(1), r.getString(2))| )
rdd: org.apache.spark.rdd.JdbcRDD[(Int, String)] &＃61; JdbcRDD[1] at JdbcRDD at :24scala> println(rdd.count())
3scala> rdd.foreach(println)
(1,zhangsan)
(2,lisi)
(3,wangwu)

4.2.3 HBase 数据库 

　　由于 org.apache.hadoop.hbase.mapreduce.TableInputFormat 类的实现&＃xff0c;Spark 可以通过

Hadoop 输入格式访问 HBase。这个输入格式会返回键值对数据&＃xff0c;其中键的类型为 org.

apache.hadoop.hbase.io.ImmutableBytesWritable&＃xff0c;而值的类型为 org.apache.hadoop.hbase.client.

Result。 

&＃xff08;1&＃xff09;添加依赖 

org.apache.hbasehbase-server1.3.1
org.apache.hbasehbase-client1.3.1

&＃xff08;2&＃xff09;从 HBase 读取数据 

package com.lxlimport org.apache.hadoop.conf.Configuration
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
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.hadoop.hbase.util.Bytesobject HBaseSpark {def main(args: Array[String]): Unit &＃61; {//创建 spark 配置信息val sparkConf: SparkConf &＃61; new SparkConf().setMaster("local[*]").setAppName("JdbcRDD")//创建 SparkContextval sc &＃61; new SparkContext(sparkConf)//构建 HBase 配置信息val conf: Configuration &＃61; HBaseConfiguration.create()conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104")conf.set(TableInputFormat.INPUT_TABLE, "rddtable")//从 HBase 读取数据形成 RDDval hbaseRDD: RDD[(ImmutableBytesWritable, Result)] &＃61; sc.newAPIHadoopRDD(conf,classOf[TableInputFormat],classOf[ImmutableBytesWritable],classOf[Result])val count: Long &＃61; hbaseRDD.count()println(count)//对 hbaseRDD 进行处理hbaseRDD.foreach {case (_, result) &＃61;>val key: String &＃61; Bytes.toString(result.getRow)val name: String &＃61; Bytes.toString(result.getValue(Bytes.toBytes("info"),Bytes.toBytes("name")))val color: String &＃61; Bytes.toString(result.getValue(Bytes.toBytes("info"),Bytes.toBytes("color")))println("RowKey:" &＃43; key &＃43; ",Name:" &＃43; name &＃43; ",Color:" &＃43; color)}//关闭连接sc.stop()}
}

3&＃xff09;往 HBase 写入 

def main(args: Array[String]) {//获取 Spark 配置信息并创建与 spark 的连接val sparkConf &＃61; new SparkConf().setMaster("local[*]").setAppName("HBaseApp")val sc &＃61; new SparkContext(sparkConf)//创建 HBaseConfval conf &＃61; HBaseConfiguration.create()val jobConf &＃61; new JobConf(conf)jobConf.setOutputFormat(classOf[TableOutputFormat[ImmutableBytesWritable]])jobConf.set(TableOutputFormat.OUTPUT_TABLE, "fruit_spark")//构建 Hbase 表描述器val fruitTable &＃61; TableName.valueOf("fruit_spark")val tableDescr &＃61; new HTableDescriptor(fruitTable)tableDescr.addFamily(new HColumnDescriptor("info".getBytes))//创建 Hbase 表val admin &＃61; new HBaseAdmin(conf)if (admin.tableExists(fruitTable)) {admin.disableTable(fruitTable)admin.deleteTable(fruitTable)}admin.createTable(tableDescr)//定义往 Hbase 插入数据的方法def convert(triple: (Int, String, Int)) &＃61; {val put &＃61; new Put(Bytes.toBytes(triple._1))put.addImmutable(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes(triple._2))put.addImmutable(Bytes.toBytes("info"), Bytes.toBytes("price"), Bytes.toBytes(triple._3))(new ImmutableBytesWritable, put)}//创建一个 RDDval initialRDD &＃61; sc.parallelize(List((1,"apple",11), (2,"banana",12), (3,"pear",13)))//将 RDD 内容写到 HBaseval localData &＃61; initialRDD.map(convert)localData.saveAsHadoopDataset(jobConf)
}