讀取結構化數據

　　Spark可以從本地CSV，HDFS以及Hive讀取結構化數據，直接解析為DataFrame，進行後續分析。

　　讀取本地CSV

　　需要指定一些選項，比如留header，比如指定delimiter值，用，或者\t或者其他。

　　import org.apache.spark.sql.{DataFrame, SparkSession}object ReadCSV { val spark: SparkSession = SparkSession

　　.builder()

　　.appName(Spark Rocks)

　　.master(local[*])

　　.getOrCreate() val path: String = /path/to/file/data.csv

　　val df: DataFrame = spark.read

　　.option(header,true)

　　.option(inferSchema,true)

　　.option(delimiter,,)

　　.csv(path)

　　.toDF() def main(args: Array[String]): Unit = {

　　df.show()

　　df.printSchema()

　　}

　　}

　　讀取Hive數據

　　SparkSession可以直接調用sql方法，傳入sql查詢語句即可。返回的DataFrame可以做簡單的變化，比如轉換

　　數據類型，對重命名之類。

　　import org.apache.spark.sql.{DataFrame, SparkSession}import org.apache.spark.sql.types.IntegerTypeobject ReadHive { val spark: SparkSession = SparkSession

　　.builder()

　　.appName(Spark Rocks)

　　.master(local[*])

　　.enableHiveSupport() // 需要開啟Hive支持

　　.getOrCreate() import spark.implicits._ //隱式轉換

　　val sql: String = SELECT col1, col2 FROM db.myTable LIMIT 1000

　　val df: DataFrame = spark.sql(sql)

　　.withColumn(col1, $col1.cast(IntegerType))

　　.withColumnRenamed(col2,new_col2) def main(args: Array[String]): Unit = {

　　df.show()

　　df.printSchema()

　　}

　　}

　　讀取HDFS數據

　　HDFS上沒有數據無法獲取表頭，需要單獨指定。可以參考databricks的網頁。一般HDFS默認在9000端口訪問。

　　import org.apache.spark.sql.{DataFrame, SparkSession}object ReadHDFS { val spark: SparkSession = SparkSession

　　.builder()

　　.appName(Spark Rocks)

　　.master(local[*])

　　.getOrCreate() val location: String = hdfs://localhost:9000/user/zhangsan/test

　　val df: DataFrame = spark

　　.read

　　.format(com.databricks.spark.csv)

　　.option(inferSchema,true)

　　.option(delimiter,\001)

　　.load(location)

　　.toDF(col1,col2) def main(args: Array[String]): Unit = {

　　df.show()

　　df.printSchema()

　　}

　　}

?

Spark讀取結構化數據