二、Spark DataFrame基礎操作

2.1、DataFrame

• DataFrame是一種不可變的分布式數據集，這種數據集被組織成指定的列，類似于關系數據庫中的表。
• 數據集的每一列都帶有名稱和類型，對于數據的內部結構又很強的描述性。
• RDD是分布式的Java對象的集合。DataFrame是分布式的Row對象的集合。

---

2.1.1、DataFrame特點

• 支持KB級到PB級得到數據處理
• 支持多種數據格式和存儲系統
• 通過Spark SQL Catalyst優化器可以進行高效的代碼生成與優化
• 能夠無縫集成Spark上的大數據處理工具
• 提供了Python、Java、Scala、R等多種語言API

---

2.2、創建DataFrame

• 從結構化數據文件創建DataFrame
• 從外部數據庫創建DataFrame
• 從RDD創建DataFrame
• 從Hive中的表創建DataFrame

2.2.1、從結構化數據文件創建DataFrame

初始化一個SparkSession，名稱為spark

spark.read.parquet(path: String)：讀取一個Parquet文件，返回一個DataFrame

spark.read.json(path: String)：讀取一個JSON文件，返回一個DataFrame

spark.read.csv(path: String)：讀取CSV文件，返回一個DataFrame

spark.read.text(path: String)：讀取文本文件的數據，返回一個DataFrame，只有一個value字段

查看DataFrame

• printSchema：查看數據模式，打印出列的名稱和類型
• show：查看數據
  show()：顯示前20條記錄
  show(numRows:Int)：顯示numRows條
  show(truncate:Boolean)：是否最多只顯示20個字符，默認為true
  show(numRows:Int,truncate:Boolean)：顯示numRows條記錄并設置過長字符串的顯示格式
• first/head/take/takeAsList：獲取若干行數據
• collect/collectAsList：獲取所有數據

示例：讀取parquet文件創建DataFrame
1、從parquet讀取數據：

val df_parquet = spark.read.parquet("/user/root/sparksql_data/users.parquet")

2、從json讀取數據：

val df_json = spark.read.json("/user/root/sparksql_data/people.json")

3、從csv讀取數據：

val df_csv = spark.read.option("header", "true").option("sep", ";").csv("/user/root/sparksql_data/people.csv")

---

2.2.2、從外部數據庫創建DataFrame

Spark SQL可以從外部數據庫（比如MySQL、Oracle等數據庫）中創建DataFrame

使用這種方式創建DataFrame需要通過JDBC連接或ODBC連接的方式訪問數據庫

spark.read.jdbc(url: String, table: String, properties: Properties)

示例：讀取mysql數據庫test的student表

---

2.2.3、從RDD創建DataFrame-方法1

利用反射機制推斷RDD模式，使用這種方式首先需要定義一個case class，因為只有case class才能被Spark隱式地轉換為DataFrame。
示例：

---

2.2.4、從RDD創建DataFrame-方法2

從原始RDD創建一個元素為Row類型的RDD

用StructType創建一個和RDD中Row的結構相匹配的Schema

通過SparkSession提供的createDataFrame方法將Schema應用到RDD上
示例：讀取people.txt創建DataFrame

---

2.2.5、從Hive中的表創建DataFrame

通過SQL查詢語句

spark.read.table(tablename)
示例：

---

任務1：讀取Hive中的學生成績表bigdata和math，查看學生信息

創建Hive表bigdata和math

導入數據到表中

讀取Hive表bigdata和math創建DataFrame

查看DataFrame

---

2.3、DataFrame常用API

示例：

---

2.3.1、自定義udf-用于SQL查詢

• import org.apache.spark.sql.functions._
• import org.apache.spark.sql.types._
• udf(f: AnyRef, dataType: DataType)
  示例：
  

2.3.2、自定義udf-用于DataFrame API

• org.apache.spark.sql.functions._
• import org.apache.spark.sql.types._
• udf(f: AnyRef, dataType: DataType)
  示例：
  

---

任務2：統計每個學生的平均成績和總分

關聯bigdata和math

根據學生ID分組統計成績平均分和總分

并根據總分進行降序排序

---

2.4、保存DataFrame

• 1、保存成文本文件

• 2、保存到外部數據庫

• 3、保存為Hive表

• write方法可以將DataFrame保存成文件，write操作有一個可選參數SaveMode，用這個參數可以指定如何處理數據已經存在的情況。

  參數SaveMode：
  Overwrite代表覆蓋目錄下之前存在的數據
  Append代表給指導目錄下追加數據
  Ignore代表如果目錄下已經有文件，那就什么都不執行
  ErrorIfExists代表如果保存目錄下存在文件就報錯

---

2.4.1、保存為文本文件

• df.write.mode("overwrite").parquet(path: String)
  

• df.write.mode("overwrite"). json(path: String)
  

• df.write.mode("overwrite").option("header", "true").option("sep", ";"). csv(path: String)
  

• df.write.mode("overwrite").text(path: String) //只能有一個字段

---

2.4.2、保存到外部數據庫

• jdbc(url: String, table: String, connectionProperties: Properties)
  
  mysql查看：
  

---

2.4.3、保存到Hive

• df.write.mode("overwrite"). saveAsTable(tableName: String)
  
  hive查看(先進入train數據庫,然后再查看)：
  

---

任務3：將每個學生各科成績、總分、平均分都保存到Hive中

• 合并學生的各科成績、總分、平均分
• 保存DataFrame到Hive
  
  hive查看(先進入train數據庫,然后再查看)：
  
  結果：
  

總結

以上是生活随笔為你收集整理的学习笔记Spark（七）—— Spark SQL应用（2）—— Spark DataFrame基础操作的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。