1、文件读取与保存
1.1、Text 文件
1）数据读取：textFile(String) 2）数据保存：saveAsTextFile(String)def main(args: Array[String]): Unit = {
//1.创建SparkConf并设置App名称
val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")
//2.创建SparkContext，该对象是提交Spark App的入口
val sc: SparkContext = new SparkContext(conf)
//3.1 读取输入文件
val inputRDD: RDD[String] = sc.textFile("input/1.txt")
//3.2 保存数据
inputRDD.saveAsTextFile("output")
//4.关闭连接
sc.stop()
}
说明：如果是集群路径：hdfs://hadoop103:9000/input/1.txt
1.2、Json 文件
Json 文件准备
{"name": "linghc","age":29}
{"name": "yilin","age":18}
{"name": "renyy","age":25}
读取保存逻辑
import com.alibaba.fastjson.JSON
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Spark05_json {
def main(args: Array[String]): Unit = {
//获取 SparkConf 并设置应用名称*本地模式
val conf: SparkConf = new SparkConf().setAppName("Spark").setMaster("local[*]")
//获取 Spark 上下文对象
val sc: SparkContext = new SparkContext(conf)
val rdd: RDD[String] = sc.textFile("E:\\workspace_idea\\spark0520\\input\\user.json")
val mapRDD: RDD[AnyRef] = rdd.map {
str => {
JSON.parse(str)
}
}
mapRDD.collect().foreach(println)
//释放 Spark 上下文对象
sc.stop
}
}
如果Json文件格式如下：
[{"name": "linghc","age":29}
{"name": "yilin","age":18}
{"name": "renyy","age":25}]
再次执行程序，发现解析失败。原因是Spark 读取Json 文件和读取Text文件是一样的，按行读取文件。
注意：使用RDD读取JSON文件处理很复杂，同时Sparksql集成了很好的处理JSON文件的方式，所以应用中多是采用Sparksql处理JSON文件。
1.3、Sequence文件
SequenceFile文件是Hadoop用来存储二进制形式的key-value对而设计的一种平面文件(Flat File)。在SparkContext中，可以调用sequenceFile[keyClass, valueClass](path)。def main(args: Array[String]): Unit = {
//1.创建SparkConf并设置App名称
val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")
//2.创建SparkContext，该对象是提交Spark App的入口
val sc: SparkContext = new SparkContext(conf)
//3.1 创建rdd
val dataRDD: RDD[(Int, Int)] = sc.makeRDD(Array((1,2),(3,4),(5,6)))
//3.2 保存数据为SequenceFile
dataRDD.saveAsSequenceFile("output")
//3.3 读取SequenceFile文件
sc.sequenceFile[Int,Int]("output").collect().foreach(println)
//4.关闭连接
sc.stop()
}
说明：SequenceFile文件只针对PairRDD
1.4、Object对象文件
对象文件是将对象序列化后保存的文件，采用Java的序列化机制。可以通过objectFile[k,v](path)函数接收一个路径，读取对象文件，返回对应的RDD，也可以通过调用saveAsObjectFile()实现对对象文件的输出。因为是序列化所以要指定类型。def main(args: Array[String]): Unit = {
//1.创建SparkConf并设置App名称
val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")
//2.创建SparkContext，该对象是提交Spark App的入口
val sc: SparkContext = new SparkContext(conf)
//3.1 创建RDD
val dataRDD: RDD[Int] = sc.makeRDD(Array(1,2,3,4))
//3.2 保存数据
dataRDD.saveAsObjectFile("output")
//3.3 读取数据
sc.objectFile[(Int)]("output").collect().foreach(println)
//4.关闭连接
sc.stop()
}
2、文件系统类数据读取与保存
2.1、HDFS
Spark的整个生态系统与Hadoop是完全兼容的，所以对于Hadoop所支持的文件类型或者数据库类型，Spark也同样支持。另外，由于Hadoop的API有新旧两个版本，所以Spark为了能够兼容Hadoop所有的版本，也提供了两套创建操作接口。对于外部存储创建操作而言，hadoopRDD和newHadoopRDD是最为抽象的两个函数接口2.2、MysqL
添加依赖
dependency>
<groupId>MysqL</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.6</version>
</dependency>
读数据
def main(args: Array[String]): Unit = {
//获取 SparkConf 并设置应用名称*本地模式
val conf: SparkConf = new SparkConf().setAppName("Spark").setMaster("local[*]")
//获取 Spark 上下文对象
val sc: SparkContext = new SparkContext(conf)
// 通过 jdbRDD 交互MysqL
/**
* sc: SparkContext, 上下文
* getConnection: () => Connection, 获取连接
* sql: String, sql
* lowerBound: Long,
* upperBound: Long,
* numPartitions: Int,分区数
* mapRow: (ResultSet) => T = JdbcRDD.resultSetToObjectArray _) 结果姐
*/
//3.定义连接MysqL的参数
val driver = "com.MysqL.jdbc.Driver"
val url = "jdbc:MysqL://hadoop103:3306/db01"
val userName = "user"
val passWd = "pwd"
var sql: String = "select * from user where id >= ? and id <= ?"
val jdbcRDD: JdbcRDD[(Int, String)] = new JdbcRDD(sc,
() => {
Class.forName(driver)
DriverManager.getConnection(url, userName, passWd)
},
sql,
0,
1,
2,
rs => {
(rs.getInt(1), rs.getString(2))
}
)
jdbcRDD.collect().foreach(println)
//释放 Spark 上下文对象
sc.stop
}
写数据
def main(args: Array[String]): Unit = {
//获取 SparkConf 并设置应用名称*本地模式
val conf: SparkConf = new SparkConf().setAppName("Spark").setMaster("local[*]")
//获取 Spark 上下文对象
val sc: SparkContext = new SparkContext(conf)
val rdd: RDD[(Int, String)] = sc.makeRDD(List((2, "linghc"), (3, "yilin")))
val driver = "com.MysqL.jdbc.Driver"
val url = "jdbc:MysqL://hadoop103:3306/db01"
val userName = "user"
val passWd = "pwd"
//以分区为单位处理
rdd.foreachPartition(datas => {
Class.forName(driver)
//创建链接
val connection: Connection = DriverManager.getConnection(url, userName, passWd)
//创建操作对象
val ps: PreparedStatement = connection.prepareStatement("insert into user (id,name) values (?,?)")
datas.foreach {
case (id, name) => {
//注册驱动
//占位符 赋值
ps.setInt(1, id)
ps.setString(2, name)
//exe
ps.executeUpdate()
}
} //释放资源
ps.close()
connection.close()
})
//释放 Spark 上下文对象
sc.stop
}
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。
