离线计算组件篇-Spark-文件读写

文件数据读写总结

• 对于RDD而言，每一次转换操作都会产生不同的RDD，供给下一个“转换”使用。
• 转换得到的RDD是惰性求值的，也就是说，整个转换过程只是记录了转换的轨迹，并不会发生真正的计算，只有遇到行动操作时，才会发生真正的计算，开始从血缘关系源头开始，进行物理的转换操作。

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)