DataFrame概述和使用-粵嵌教育

更新時間： 2018-09-06 18:30:26來源： java大數據瀏覽量：6689

一、概述：
DataFrame是一個分布式數據集,可以理解為關系型數據庫一張表，由字段和字段類型、字段值按列組織，且支持四種語言，在Scala API中可以理解為： FataFrame=Dataset[ROW]
注：DataFrame產生于V1.3之后，在V1.3前為SchemaRDD,在V1.6以后又添加了Dataset

二、DataFrame vs RDD 差異：

概念 :


兩個都是分布式容器，DF理解是一個表格除了RDD數據以外還有Schema，也支持復雜數據類型（map..）
 
API :


DataFrame提供的API比RDD豐富支持map filter flatMap .....
 
數據結構：RDD知道類型沒有結構， DF提供Schema信息有利于優化,性能上好
 
底層：基于運行環境不一樣,RDD開發的Java/Scala API運行底層環境JVM,
 


DF在SparkSQL中轉換成邏輯執行計劃(locaical Plan)和物理執行計劃(Physical Plan)中間自身優化功能，性能差異大

三、json文件操作

[hadoop@hadoop001 bin]$./spark-shell --master local[2] --jars ~/software/mysql-connector-java-5.1.34-bin.jar
-- 讀取json文件
scala>val df = spark.read.json("file:///home/hadoop/data/people.json")
18/09/02 11:47:20 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
-- 打印schema信息

scala> df.printSchema

root
 
|-- age: long (nullable = true) -- 字段類型允許為空
 
|-- name: string (nullable = true)
 

-- 打印字段內容

scala> df.show

+----+-------+
 
| age| name|
 
+----+-------+
 
|null|Michael|
 
| 30| Andy|
 
| 19| Justin|
 
+----+-------+

-- 打印查詢字段

scala> df.filter($"age" > 21).show

+---+----+
 
|age|name|
 
+---+----+
 
| 30|Andy|
 
+---+----+
 

-- 年齡分組匯總

scala> df.groupBy("age").count.show

+----+-----+
 
| age|count|
 
+----+-----+
 
| 19| 1|
 
|null| 1|
 
| 30| 1|
 
+----+-----+
 

-- 創建一個臨時視圖
scala> df.createOrReplaceTempView("people")

scala>spark.sql("select * from people").show

+----+-------+
 
| age| name|
 
+----+-------+
 
|null|Michael|
 
| 30| Andy|
 
| 19| Justin|
 
+----+-------+

四、DataFrame對象上Action操作

-- 定義case class 用來創建Schema
case class Student(id:String,name:String,phone:String,Email:String)
-- RDD與DF反射方式實現
val students = sc.textFile("file:///home/hadoop/data/student.data").map(_.split("\|")).map(x=>Student(x(0),x(1),x(2),x(3))).toDF()
-- 打印DF信息
students.printSchema
-- show(numRows: Int, truncate: Boolean)
-- numRows截取前20行和truncate讀取前20字符串
-- students.show(5,false) 讀取前五行和所有字符串
scala> students.show

+---+--------+--------------+--------------------+
 
| id| name| phone| Email|
 
+---+--------+--------------+--------------------+
 
| 1| Burke|1-300-746-8446|ullamcorper.velit...|
 
| 2| Kamal|1-668-571-5046|pede.Suspendisse@...|
 
| 3| Olga|1-956-311-1686|Aenean.eget.metus...|
 
| 4| Belle|1-246-894-6340|vitae.aliquet.nec...|
 
| 5| Trevor|1-300-527-4967|dapibus.id@acturp...|
 
| 6| Laurel|1-691-379-9921|adipiscing@consec...|
 
| 7| Sara|1-608-140-1995|Donec.nibh@enimEt...|
 
| 8| Kaseem|1-881-586-2689|cursus.et.magna@e...|
 
| 9| Lev|1-916-367-5608|Vivamus.nisi@ipsu...|
 
| 10| Maya|1-271-683-2698|accumsan.convalli...|
 
| 11| Emi|1-467-270-1337|est@nunc.com|.......|
 
| 12| Caleb|1-683-212-0896|Suspendisse@Quisq...|
 
| 13|Florence|1-603-575-2444|sit.amet.dapibus@...|
 
| 14| Anika|1-856-828-7883|euismod@ligulaeli...|
 
| 15| Tarik|1-398-171-2268|turpis@felisorci.com|
 
| 16| Amena|1-878-250-3129|lorem.luctus.ut@s...|
 
| 17| Blossom|1-154-406-9596|Nunc.commodo.auct...|
 
| 18| Guy|1-869-521-3230|senectus.et.netus...|
 
| 19| Malachi|1-608-637-2772|Proin.mi.Aliquam@...|
 
| 20| Edward|1-711-710-6552|lectus@aliquetlib...|
 
+---+--------+--------------+--------------------+
 
only showing top 20 rows

-- students.head(5) 返回前幾行數據

scala> students.head(5).foreach(println)
 
[1,Burke,1-300-746-8446,ullamcorper.velit.in@ametnullaDonec.co.uk]
 
[2,Kamal,1-668-571-5046,pede.Suspendisse@interdumenim.edu]
 
[3,Olga,1-956-311-1686,Aenean.eget.metus@dictumcursusNunc.edu]
 
[4,Belle,1-246-894-6340,vitae.aliquet.nec@neque.co.uk]
 
[5,Trevor,1-300-527-4967,dapibus.id@acturpisegestas.net]
 

-- 查詢具體字段

scala> students.select("id","name").show(5)
 
+---+------+
 
| id| name|
 
+---+------+
 
| 1| Burke|
 
| 2| Kamal|
 
| 3| Olga|
 
| 4| Belle|
 
| 5|Trevor|
 
+---+------+
 

-- 修改字段取別名
scala> students.select($"name".as("new_name")).show(5)

+---+------+--------------+--------------------+
 
| id| name| phone| Email|
 
+---+------+--------------+--------------------+
 
| 6|Laurel|1-691-379-9921|adipiscing@consec...|
 
| 7| Sara|1-608-140-1995|Donec.nibh@enimEt...|
 
| 8|Kaseem|1-881-586-2689|cursus.et.magna@e...|
 
| 9| Lev|1-916-367-5608|Vivamus.nisi@ipsu...|
 
| 10| Maya|1-271-683-2698|accumsan.convalli...|
 
+---+------+--------------+--------------------+
 

-- 查詢名稱為空或者名稱為NULL(filter=where)
scala> students.filter("name=''or name='NULL'").show(false)

+---+----+--------------+--------------------------+
 
|id |name|phone |Email |
 
+---+----+--------------+--------------------------+
 
|21 | |1-711-710-6552|lectus@aliquetlibero.co.uk|
 
|22 | |1-711-710-6552|lectus@aliquetlibero.co.uk|
 
|23 |NULL|1-711-710-6552|lectus@aliquetlibero.co.uk|
 
+---+----+--------------+--------------------------+
 

-- 查詢ID大于5且名稱模糊查詢
scala> students.filter("id>5 and name like 'M%'").show(5)

+---+-------+--------------+--------------------+
 
| id| name| phone| Email|
 
+---+-------+--------------+--------------------+
 
| 10| Maya|1-271-683-2698|accumsan.convalli...|
 
| 19|Malachi|1-608-637-2772|Proin.mi.Aliquam@...|
 
+---+-------+--------------+--------------------+
 

-- 按照名稱升序排序且不等于空
scala> students.sort($"name").select("id","name").filter("name <> ''").show(3)

+---+-----+
 
| id| name|
 
+---+-----+
 
| 16|Amena|
 
| 14|Anika|
 
| 4|Belle|
 
+---+-----+
 

-- 按照名稱倒敘排序(sort = orderBy)
scala> students.sort($"name".desc).select("name").show(5)

+----+-----+
 
| age|count|
 
+----+-----+
 
| 19| 1|
 
|null| 1|
 
| 30| 1|
 
+----+-----+
 

-- 聚合函數使用
scala> students.agg("id" -> "max", "id" -> "sum").show(false)

+-------+-------+
 
|max(id)|sum(id)|
 
+-------+-------+
 
|9 |276.0 |
 
+-------+-------+
 

-- join操作,using模式seq指定多個字段

students.join(students2, Seq("id", "name"), "inner"）
 

-- DataFrame的join操作有inner, outer, left_outer, right_outer, leftsemi類型
-- 指定類型，指定join的類型

students.join(students2 , students("id" ) === students2( "t1_id"), "inner")

五、DataFrame API實現文件操作
1.maven依賴下載

<spark.version>2.3.1</spark.version>
 
 

 
<dependency>
 
<groupId>org.apache.spark</groupId>
 
<artifactId>spark-core_2.11</artifactId>
 
<version>${spark.version}</version>
 
</dependency>
 
 

 
<dependency>
 
<groupId>org.apache.spark</groupId>
 
<artifactId>spark-sql_2.11</artifactId>
 
<version>${spark.version}</version>
 
</dependency>

2、IDEA實現方式：

package com.zrc.ruozedata.sparkSQL
 
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
 
import org.apache.spark.sql.{Row, SparkSession}
 
 
object SparkSQL001 extends App {
 
/*
 
* RDD與DataFrame反射方式實現（一）
 
* 創建RDD --> DataFrema
 
* 利用case class創建Schema,來解析輸出文本每一行信息
 
*/
 
val spark = SparkSession.builder()
 
.master("local[2]")
 
.appName("SparkSQL001")
 
.getOrCreate() // 操作hive添加
 
val infos = spark.sparkContext.textFile("file:///F:/infos.txt")
 
 
/*
 
import spark.implicits._
 
val infoDF = infos.map(_.split(",")).map(x=>Info(x(0).toInt,x(1),x(2).toInt)).toDF()
 
infoDF.show()
 
*/
 
 
/*
 
* RDD與DataFrame使用StructType方式實現(二)
 
* StructType構造了StructField方法傳入name和dataType
 
* 每一個字段就是為一個StructField
 
* Schema和RDD通過createDataFrame方法作用起來
 
*/
 
// 注意通過ROW獲取的需要轉換對應類型
 
val infoss = infos.map(_.split(",")).map(x=>Row(x(0).trim.toInt,x(1),x(2).trim.toInt))
 
val fields = StructType(
 
Array(
 
StructField("id",IntegerType,true),
 
StructField("name",StringType,true),
 
StructField("age",IntegerType,true)
 
)
 
)
 
val schema = StructType(fields)
 
val infoDF = spark.createDataFrame(infoss,schema)
 
infoDF.show()
 
spark.stop()
 
}
 
// case class Info (id:Int,name:String,age:Int)

上一篇：java大數據幫你把shell-tr命令詳解-粵嵌培訓

下一篇：Tensorflow學習: 變量及共享變量-Python人工智能

DataFrame概述和使用-粵嵌教育

免費預約試聽課

粵嵌動態

DataFrame概述和使用-粵嵌教育

免費預約試聽課

粵嵌動態

推薦閱讀