package com.profile.main
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
import org.apache.log4j.{Level, Logger}
import com.profile.tools.{DateTools, JdbcTools, LogTools, SparkTools}
import com.dhd.comment.Constant
import com.profile.comment.Comments
/**
* 測試類 //使用dataframe解決spark TopN問題：分組、排序、取TopN
* @author
* date 2017-09-27 14:55
*/
object Test {
def main(args: Array[String]): Unit = {
val sc=SparkTools.getSparkContext
//設置日志級別，避免輸出大量無關的信息
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark.sql").setLevel(Level.WARN)
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext.implicits._
1、使用dataframe解決spark TopN問題
val df = sc.parallelize(Seq(
(0,"cat26",30.9), (0,"cat13",22.1), (0,"cat95",19.6), (0,"cat105",1.3),
(1,"cat67",28.5), (1,"cat4",26.8), (1,"cat13",12.6), (1,"cat23",5.3),
(2,"cat56",39.6), (2,"cat40",29.7), (2,"cat187",27.9), (2,"cat68",9.8),
(3,"cat8",35.6))).toDF("Hour", "Category", "TotalValue")
df.show
/*
+----+--------+----------+
|Hour|Category|TotalValue|
+----+--------+----------+
| 0| cat26| 30.9|
| 0| cat13| 22.1|
| 0| cat95| 19.6|
| 0| cat105| 1.3|
| 1| cat67| 28.5|
| 1| cat4| 26.8|
| 1| cat13| 12.6|
| 1| cat23| 5.3|
| 2| cat56| 39.6|
| 2| cat40| 29.7|
| 2| cat187| 27.9|
| 2| cat68| 9.8|
| 3| cat8| 35.6|
+----+--------+----------+
*/
/* val w = Window.partitionBy($"Hour").orderBy($"TotalValue".desc)
//取Top1
val dfTop1 = df.withColumn("rn", rowNumber.over(w)).where($"rn" === 1).drop("rn")
//注意：row_number()在spark1.x版本中為rowNumber(),在2.x版本為row_number()
//取Top3
val dfTop3 = df.withColumn("rn", rowNumber.over(w)).where($"rn" <= 3).drop("rn")
dfTop1.show*/
/*
+----+--------+----------+
|Hour|Category|TotalValue|
+----+--------+----------+
| 1| cat67| 28.5|
| 3| cat8| 35.6|
| 2| cat56| 39.6|
| 0| cat26| 30.9|
+----+--------+----------+
*/
// dfTop3.show
/*
+----+--------+----------+
|Hour|Category|TotalValue|
+----+--------+----------+
| 1| cat67| 28.5|
| 1| cat4| 26.8|
| 1| cat13| 12.6|
| 3| cat8| 35.6|
| 2| cat56| 39.6|
| 2| cat40| 29.7|
| 2| cat187| 27.9|
| 0| cat26| 30.9|
| 0| cat13| 22.1|
| 0| cat95| 19.6|
+----+--------+----------+
*/
//使用RDD解決spark TopN問題：分組、排序、取TopN
val rdd1 = sc.parallelize(Seq(
(0,"cat26",30.9), (0,"cat13",22.1), (0,"cat95",19.6), (0,"cat105",1.3),
(1,"cat67",28.5), (1,"cat4",26.8), (1,"cat13",12.6), (1,"cat23",5.3),
(2,"cat56",39.6), (2,"cat40",29.7), (2,"cat187",27.9), (2,"cat68",9.8),
(3,"cat8",35.6)))
val rdd2 = rdd1.map(x => (x._1,(x._2, x._3))).groupByKey()
/*
rdd2.collect
res9: Array[(Int, Iterable[(String, Double)])] = Array((0,CompactBuffer((cat26,30.9), (cat13,22.1), (cat95,19.6), (cat105,1.3))),
(1,CompactBuffer((cat67,28.5), (cat4,26.8), (cat13,12.6), (cat23,5.3))),
(2,CompactBuffer((cat56,39.6), (cat40,29.7), (cat187,27.9), (cat68,9.8))),(3,CompactBuffer((cat8,35.6))))
*/
val N_value = 1 //取前3
val rdd3 = rdd2.map( x => {
val i2 = x._2.toBuffer
val i2_2 = i2.sortBy(_._2)
if (i2_2.length > N_value) i2_2.remove(0, (i2_2.length - N_value))
(x._1, i2_2.toIterable)
})
/*
rdd3.collect
res8: Array[(Int, Iterable[(String, Double)])] = Array((0,ArrayBuffer((cat95,19.6), (cat13,22.1), (cat26,30.9))),
(1,ArrayBuffer((cat13,12.6), (cat4,26.8), (cat67,28.5))),
(2,ArrayBuffer((cat187,27.9), (cat40,29.7), (cat56,39.6))),(3,ArrayBuffer((cat8,35.6))))
*/
val rdd4 = rdd3.flatMap(x => {
val y = x._2
for (w <- y) yield (x._1, w._1, w._2)
})
rdd4.collect
/*
res3: Array[(Int, String, Double)] = Array((0,cat95,19.6), (0,cat13,22.1), (0,cat26,30.9),
(1,cat13,12.6), (1,cat4,26.8), (1,cat67,28.5),
(2,cat187,27.9), (2,cat40,29.7), (2,cat56,39.6), (3,cat8,35.6))
*/
rdd4.toDF("Hour", "Category", "TotalValue").show
/* +----+--------+----------+
|Hour|Category|TotalValue|
+----+--------+----------+
| 0| cat95| 19.6|
| 0| cat13| 22.1|
| 0| cat26| 30.9|
| 2| cat187| 27.9|
| 2| cat40| 29.7|
| 2| cat56| 39.6|
| 1| cat13| 12.6|
| 1| cat4| 26.8|
| 1| cat67| 28.5|
| 3| cat8| 35.6|
+----+--------+----------+*/
}
2、下面再來看DataFrame的join操作。
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
/**
* spark-DataFrame學習記錄-[2]解決spark-dataframe的JOIN操作之后產生重復列（Reference '***' is ambiguous問題解決）
*/
object DataFrameSQL_2 {
def main(args: Array[String]) {
val conf = new SparkConf()
conf.setAppName("test").setMaster("local")
val sc = new SparkContext(conf)
//設置日志級別
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark.sql").setLevel(Level.WARN)
val sqlContext = new HiveContext(sc)
import sqlContext.implicits._
val df = sc.parallelize(Array(
("one", "A", 1), ("one", "B", 2), ("two", "A", 3), ("two", "B", 4)
)).toDF("key1", "key2", "value")
df.show()
// +----+----+-----+
// |key1|key2|value|
// +----+----+-----+
// | one| A| 1|
// | one| B| 2|
// | two| A| 3|
// | two| B| 4|
// +----+----+-----+
val df2 = sc.parallelize(Array(
("one", "A", 5), ("two", "A", 6)
)).toDF("key1", "key2", "value2")
df2.show()
// +----+----+------+
// |key1|key2|value2|
// +----+----+------+
// | one| A| 5|
// | two| A| 6|
// +----+----+------+
//多列join可以用&&或者and,對其進行JOIN操作之后，發現多產生了KEY1和KEY2這樣的兩個字段
val joined = df.join(df2, df("key1") === df2("key1") && df("key2") === df2("key2"), "left_outer")
//這樣也可
//val joined = df.join(df2, df("key1") === df2("key1") and df("key2") === df2("key2"), "left_outer")
joined.show()
// +----+----+-----+----+----+------+
// |key1|key2|value|key1|key2|value2|
// +----+----+-----+----+----+------+
// | two| A| 3| two| A| 6|
// | two| B| 4|null|null| null|
// | one| A| 1| one| A| 5|
// | one| B| 2|null|null| null|
// +----+----+-----+----+----+------+
- 假如這兩個字段同時存在，那么就會報錯，如下：org.apache.spark.sql.AnalysisException: Reference 'key2' is ambiguous
- 因此，如何在JOIN之后刪除列的，可以通過修改JOIN的表達式，完全可以避免這個問題。主要是通過Seq這個對象來實現。
//多列join最好還是使用Seq()
df.join(df2, Seq("key1", "key2"), "left_outer").show()
// +----+----+-----+------+
// |key1|key2|value|value2|
// +----+----+-----+------+
// | two| A| 3| 6|
// | two| B| 4| null|
// | one| A| 1| 5|
// | one| B| 2| null|
// +----+----+-----+------+
//多列列名相同，則可以修改列名
val df22 = df2.withColumnRenamed("key1","k1").withColumnRenamed("key2","k2")
df.join(df22,df("key1") === df22("k1") && df("key2") === df22("k2"), "left_outer").show()
// +----+----+-----+----+----+------+
// |key1|key2|value| k1| k2|value2|
// +----+----+-----+----+----+------+
// | two| A| 3| two| A| 6|
// | two| B| 4|null|null| null|
// | one| A| 1| one| A| 5|
// | one| B| 2|null|null| null|
// +----+----+-----+----+----+------+
sc.stop()
}
}
3、下面再來看RDD的join操作。
/建立一個基本的鍵值對RDD，包含ID和名稱，其中ID為1、2、3、4
val Arrayrdd1 = sc.makeRDD(Array(("1","Spark"),("2","Hadoop"),("3","Scala"),("4","Java")),2)
//建立一個行業薪水的鍵值對RDD，包含ID和薪水，其中ID為1、2、3、5
val Arrayrdd2 = sc.makeRDD(Array(("1","30K"),("2","15K"),("3","25K"),("5","10K")),2)
println("//下面做Join操作，預期要得到（1,×）、（2,×）、（3,×）")
val joinRDD=Arrayrdd1.join(Arrayrdd2).collect.foreach(println)
println("//下面做leftOutJoin操作，預期要得到（1,×）、（2,×）、（3,×）、(4,×）")
val leftJoinRDD=Arrayrdd1.leftOuterJoin(Arrayrdd2).collect.foreach(println)
println("//下面做rightOutJoin操作，預期要得到（1,×）、（2,×）、（3,×）、(5,×）")
val rightJoinRDD=Arrayrdd1.rightOuterJoin(Arrayrdd2).collect.foreach(println)
結果輸出如下：
和預期的效果一樣。
4、DataFrame 的函數
Action 操作
1、 collect() ,返回值是一個數組，返回dataframe集合所有的行
2、 collectAsList() 返回值是一個java類型的數組，返回dataframe集合所有的行
3、 count() 返回一個number類型的，返回dataframe集合的行數
4、 describe(cols: String*) 返回一個通過數學計算的類表值(count, mean, stddev, min, and max)，這個可以傳多個參數，中間用逗號分隔，如果有字段為空，那么不參與運算，只這對數值類型的字段。
例如df.describe("age", "height").show()
5、 first() 返回第一行 ，類型是row類型
6、 head() 返回第一行 ，類型是row類型
7、 head(n:Int)返回n行 ，類型是row 類型
8、 show()返回dataframe集合的值 默認是20行，返回類型是unit
9、 show(n:Int)返回n行，，返回值類型是unit
10、 table(n:Int) 返回n行 ，類型是row 類型
dataframe的基本操作
1、 cache()同步數據的內存
2、 columns 返回一個string類型的數組，返回值是所有列的名字
3、 dtypes返回一個string類型的二維數組，返回值是所有列的名字以及類型
4、 explan()打印執行計划 物理的
5、 explain(n:Boolean) 輸入值為 false 或者true ，返回值是unit 默認是false ，如果輸入true 將會打印 邏輯的和物理的
6、 isLocal 返回值是Boolean類型，如果允許模式是local返回true 否則返回false
7、 persist(newlevel:StorageLevel) 返回一個dataframe.this.type 輸入存儲模型類型
8、 printSchema() 打印出字段名稱和類型 按照樹狀結構來打印
9、 registerTempTable(tablename:String) 返回Unit ，將df的對象只放在一張表里面，這個表隨着對象的刪除而刪除了
10、 schema 返回structType 類型，將字段名稱和類型按照結構體類型返回
11、 toDF()返回一個新的dataframe類型的
12、 toDF(colnames：String*)將參數中的幾個字段返回一個新的dataframe類型的，
13、 unpersist() 返回dataframe.this.type 類型，去除模式中的數據
14、 unpersist(blocking:Boolean)返回dataframe.this.type類型 true 和unpersist是一樣的作用false 是去除RDD
集成查詢：
1、 agg(expers:column*) 返回dataframe類型 ，同數學計算求值
df.agg(max("age"), avg("salary"))
df.groupBy().agg(max("age"), avg("salary"))
2、 agg(exprs: Map[String, String]) 返回dataframe類型 ，同數學計算求值 map類型的
df.agg(Map("age" -> "max", "salary" -> "avg"))
df.groupBy().agg(Map("age" -> "max", "salary" -> "avg"))
3、 agg(aggExpr: (String, String), aggExprs: (String, String)*) 返回dataframe類型 ，同數學計算求值
df.agg(Map("age" -> "max", "salary" -> "avg"))
df.groupBy().agg(Map("age" -> "max", "salary" -> "avg"))
4、 apply(colName: String) 返回column類型，捕獲輸入進去列的對象
5、 as(alias: String) 返回一個新的dataframe類型，就是原來的一個別名
6、 col(colName: String) 返回column類型，捕獲輸入進去列的對象
7、 cube(col1: String, cols: String*) 返回一個GroupedData類型，根據某些字段來匯總
8、 distinct 去重 返回一個dataframe類型
9、 drop(col: Column) 刪除某列 返回dataframe類型
10、 dropDuplicates(colNames: Array[String]) 刪除相同的列 返回一個dataframe
11、 except(other: DataFrame) 返回一個dataframe，返回在當前集合存在的在其他集合不存在的
12、 explode[A, B](inputColumn: String, outputColumn: String)(f: (A) ⇒ TraversableOnce[B])(implicit arg0: scala.reflect.api.JavaUniverse.TypeTag[B]) 返回值是dataframe類型，這個 將一個字段進行更多行的拆分
df.explode("name","names") {name :String=> name.split(" ")}.show();
將name字段根據空格來拆分，拆分的字段放在names里面
13、 filter(conditionExpr: String): 刷選部分數據，返回dataframe類型 df.filter("age>10").show(); df.filter(df("age")>10).show(); df.where(df("age")>10).show(); 都可以
14、 groupBy(col1: String, cols: String*) 根據某寫字段來匯總返回groupedate類型 df.groupBy("age").agg(Map("age" ->"count")).show();df.groupBy("age").avg().show();都可以
15、 intersect(other: DataFrame) 返回一個dataframe，在2個dataframe都存在的元素
16、 join(right: DataFrame, joinExprs: Column, joinType: String)
一個是關聯的dataframe，第二個關聯的條件，第三個關聯的類型：inner, outer, left_outer, right_outer, leftsemi
df.join(ds,df("name")===ds("name") and df("age")===ds("age"),"outer").show();
17、 limit(n: Int) 返回dataframe類型 去n 條數據出來
18、 na: DataFrameNaFunctions ，可以調用dataframenafunctions的功能區做過濾 df.na.drop().show(); 刪除為空的行
19、 orderBy(sortExprs: Column*) 做alise排序
20、 select(cols:string*) dataframe 做字段的刷選 df.select($"colA", $"colB" + 1)
21、 selectExpr(exprs: String*) 做字段的刷選 df.selectExpr("name","name as names","upper(name)","age+1").show();
22、 sort(sortExprs: Column*) 排序 df.sort(df("age").desc).show(); 默認是asc
23、 unionAll(other:Dataframe) 合並 df.unionAll(ds).show();
24、 withColumnRenamed(existingName: String, newName: String) 修改列表 df.withColumnRenamed("name","names").show();
25、 withColumn(colName: String, col: Column) 增加一列 df.withColumn("aa",df("name")).show();
}