Spark讀取結構化數據

讀取結構化數據

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

讀取本地CSV

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

import org.apache.spark.sql.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.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.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,01)

.load(location)

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

df.show()

df.printSchema()

}

}

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！