文件数据读写总结
- 对于RDD而言，每一次转换操作都会产生不同的RDD，供给下一个“转换”使用。
- 转换得到的RDD是惰性求值的，也就是说，整个转换过程只是记录了转换的轨迹，并不会发生真正的计算，只有遇到行动操作时，才会发生真正的计算，开始从血缘关系源头开始，进行物理的转换操作。
| 转换 | 含义 |
|---|---|
| map(func) | 返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成 |
| filter(func) | 返回一个新的RDD，该RDD由经过func函数(在func函数中添加_.contains()来添加过滤条件)计算后返回值为true的输入元素组成 |
| flatMap(func) | 类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素） |
| mapPartitions(func) | 类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U] |
| mapPartitionsWithIndex(func) | 类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是(Int, Interator[T]) => Iterator[U] |
| union(otherDataset) | 对源RDD和参数RDD求并集后返回一个新的RDD（不去重） |
| intersection(otherDataset) | 对源RDD和参数RDD求交集后返回一个新的RDD |
| distinct([numTasks])) | 对源RDD进行去重后返回一个新的RDD |
| groupByKey([numTasks]) | 在一个(K,V)的RDD上调用，返回一个(K, Iterator[V])的RDD |
| reduceByKey(func, [numTasks]) | 在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，与groupByKey类似，reduce任务的个数可以通过第二个可选的参数来设置 |
| sortByKey([ascending], [numTasks]) | 在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD |
| sortBy(func,[ascending], [numTasks]) | 与sortByKey类似，但是更灵活 |
| join(otherDataset, [numTasks]) | 在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD |
| cogroup(otherDataset, [numTasks]) | 在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD |
| coalesce(numPartitions) | 减少 RDD 的分区数到指定值。 |
| repartition(numPartitions) | 重新给 RDD 分区 |
| repartitionAndSortWithinPartitions(partitioner) | 重新给 RDD 分区，并且每个分区内以记录的 key 排序 |
1),本地文件系统的数据读写
- 读取本地文件
// 因为Spark采用了惰性机制，在执行转换操作的时候，即使输入了错误的语句，spark-shell也不会马上报错（假设word123.txt不存在）
scala>val textFile = sc.textFile("file:///usr/local/spark/mycode/wordcount/word.txt")
// 执行行动操作，查看数据集的第一个元素，才会真正的执行。
textFile.first()
- 写入本地文件
scala> val textFile = sc.textFile("file:///usr/local/spark/mycode/wordcount/word.txt")
// 写入文件系统的指定目录中
scala> textFile.saveAsTextFile("file:///usr/local/spark/mycode/wordcount/writeback")
// 查看文件
cd /usr/local/spark/mycode/wordcount/writeback/
ls
// 文件结果
part-r-00000
_SUCCESS
2),分布式文件向HDFS的数据读写
- 读取HDFS文件系统
从分布式文件系统HDFS中读取数据，也是采用textFile()方法，可以为textFile()方法提供一个HDFS文件或目录地址，如果是一个文件地址，它会加载该文件，如果是一个目录地址，它会加载该目录下的所有文件的数据
// 调用读取文件的方法
scala> val textFile = sc.textFile("hdfs://localhost:9000/user/hadoop/word.txt")
// 查看数据的第一个数据
scala> textFile.first()
// 以下三种方式是等价的，推荐使用第二种，因为已经在hdfs的配置文件中指定了连接信息
scala> val textFile = sc.textFile("hdfs://localhost:9000/user/hadoop/word.txt")
scala> val textFile = sc.textFile("/user/hadoop/word.txt")
scala> val textFile = sc.textFile("word.txt") // 只是在当前登录用户的家目录下
- 写入HDFS文件系统
// 从HDFS中读取文件
scala> val textFile = sc.textFile("hdfs://localhost:9000/mycode/wordcount/word.txt")
// 写入HDFS文件系统的指定目录中
// 方法1：
scala> textFile.saveAsTextFile("hdfs://localhost:9000/user/hadoop/writeback")
// 方法2：
scala> textFile.saveAsTextFile("/user/hadoop/writeback")
3),JSON文件的数据读写
JSON(JavaScriptObject Notation) 是一种轻量级的数据交换格式
Spark提供了一个JSON样例数据文件，存放在“/usr/local/spark/examples/src/main/resources/people.json”中；启动usr/local/spark是spark的安装目录
{"name":"Michael"} {"name":"Andy", "age":30} {"name":"Justin", "age":19}
- 加载本地文件系统直中的Json文件
scala> val jsonStr = sc.textFile("file:///root/wordCount/people.json")
scala> jsonStr.foreach(println)
{"name":"Michael"}
{"name":"Andy", "age":30}
{"name":"Justin", "age":19}
- 解析Json数据
- Scala中有一个自带的JSON库——scala.util.parsing.json.JSON，可以实现对JSON数据的解析
- JSON.parseFull(jsonString:String)函数，以一个JSON字符串作为输入并进行解析，如果解析成功则返回一个Some(map:Map[String, Any])，如果解析失败则返回None
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import scala.util.parsing.json.JSON
object JSONRead {
def main(args: Array[String]) {
val inputFile = "file:///usr/local/spark/examples/src/main/resources/people.json"
val conf = new SparkConf().setAppName("JSONRead")
val sc = new SparkContext(conf)
// 加载出来的数据是一个RDD
val jsonStrs = sc.textFile(inputFile)
// 解析之后，得到的是Option 类型。有值的时候是Some类型；没有值的时候是None类型
val result = jsonStrs.map(s => JSON.parseFull(s))
/*
1.依次匹配每一个元素，判读那类型是不是Some类型
*/
result.foreach( {r => r match {
case Some(map: Map[String, Any]) => println(map)
case None => println("Parsing failed")
case other => println("Unknown data structure: " + other)
}
}
)
}
}
- 打包、运行
- 使用之前的普通方式打包、或者使用Maven进行打包。
- 通过 spark-submit 运行程序
spark-submit \ --class "JSONRead" \ /usr/local/spark/mycode/json/target/scala-2.11/json-project_2.11-1.0.jar# 输出结果 Map(name -> Michael) Map(name -> Andy, age -> 30.0) Map(name -> Justin, age -> 19.0)
网友评论