常見應用 - SparkSql 之 Column類

1.什么是Column對象

Column 表示了 Dataset 中的一個列, 並且可以持有一個表達式, 這個表達式作用於每一條數據, 對每條數據都生成一個值

2.Column對象如何創建

(1)’

單引號 ’ 在 Scala 中是一個特殊的符號, 通過 ’ 會生成一個 Symbol 對象, Symbol 對象可以理解為是一個字符串的變種, 但是比字符串的效率高很多, 在 Spark 中, 對 Scala 中的 Symbol 對象做了隱式轉換, 轉換為一個 ColumnName 對象, ColumnName 是 Column 的子類, 所以在 Spark 中可以如下去選中一個列

val spark = SparkSession.builder().appName("column").master("local[6]").getOrCreate() import spark.implicits._ val personDF = Seq(Person("zhangsan", 12), Person("zhangsan", 8), Person("lisi", 15)).toDS() val c1: Symbol = 'name 

(2)$

同理, $ 符號也是一個隱式轉換, 同樣通過 spark.implicits 導入, 通過 $ 可以生成一個 Column 對象

val spark = SparkSession.builder().appName("column").master("local[6]").getOrCreate() import spark.implicits._ val personDF = Seq(Person("zhangsan", 12), Person("zhangsan", 8), Person("lisi", 15)).toDS() val c2: ColumnName = $"name" 

(3)col

SparkSQL 提供了一系列的函數, 可以通過函數實現很多功能, 在后面課程中會進行詳細介紹, 這些函數中有兩個可以幫助我們創建 Column 對象, 一個是 col, 另外一個是 column

val spark = SparkSession.builder().appName("column").master("local[6]").getOrCreate() import org.apache.spark.sql.functions._ val personDF = Seq(Person("zhangsan", 12), Person("zhangsan", 8), Person("lisi", 15)).toDS() val c3: sql.Column = col("name") 

(4)column

val spark = SparkSession.builder().appName("column").master("local[6]").getOrCreate() import org.apache.spark.sql.functions._ val personDF = Seq(Person("zhangsan", 12), Person("zhangsan", 8), Person("lisi", 15)).toDS() val c4: sql.Column = column("name") 

(5)Dataset.col

面的 Column 對象創建方式所創建的 Column 對象都是 Free 的, 也就是沒有綁定任何 Dataset, 所以可以作用於任何 Dataset, 同時, 也可以通過 Dataset 的 col 方法選擇一個列, 但是這個 Column 是綁定了這個 Dataset 的, 所以只能用於創建其的 Dataset 上

val spark = SparkSession.builder().appName("column").master("local[6]").getOrCreate() val personDF = Seq(Person("zhangsan", 12), Person("zhangsan", 8), Person("lisi", 15)).toDS() val c5: sql.Column = personDF.col("name") 

(6)Dataset.apply

可以通過 Dataset 對象的 apply 方法來獲取一個關聯此 Dataset 的 Column 對象
ds(“name”)
ds.apply(“name”) 上下兩個是一樣的,ds(“name”)其實是ds.apply(“name”)簡寫版

val spark = SparkSession.builder().appName("column").master("local[6]").getOrCreate() val personDF = Seq(Person("zhangsan", 12), Person("zhangsan", 8), Person("lisi", 15)).toDS() val c6: sql.Column = personDF.apply("name") apply 的調用有一個簡寫形式 val c7: sql.Column = personDF("name") 

3.別名和轉換

(1)as[type]

as 方法有兩個用法, 通過 as[Type] 的形式可以將一個列中數據的類型轉為 Type 類型

personDF.select(col("age").as[Long]).show() 

(2)as(name)

通過 as(name) 的形式使用 as 方法可以為列創建別名

personDF.select(col("age").as("age_new")).show() 

4.添加列

(1)withColumn

通過 Column 在添加一個新的列時候修改 Column 所代表的列的數據

personDF.withColumn("double_age", 'age * 2).show() 

5.操作

(1)like

通過 Column 的 API, 可以輕松實現 SQL 語句中 LIKE 的模糊查詢功能

personDF.filter('name like "%zhang%").show() //%通配符 

(2)isin

通過 Column 的 API, 可以輕松實現 SQL 語句中 ISIN 的枚舉判斷功能

personDF.filter('name isin ("hello", "zhangsan")).show()//在name字段下我只想看hello,zhangsan的信息 

(3)sort

在排序的時候, 可以通過 Column 的 API 實現正反序

personDF.sort('age.asc).show() personDF.sort('age.desc).show()

(4)lit()

使用lit()增加常量（固定值）

scala> df.withColumn("sinurl", lit(12)).show 
+----+----+----+------+
|  id|rsrp|rsrq|sinurl|
+----+----+----+------+
|key1|  23| 1.0|    12|
|key1|  10| 2.0|    12|
+----+----+----+------+

scala> df.withColumn("type", lit("mr")).show 
+----+----+----+----+
|  id|rsrp|rsrq|type|
+----+----+----+----+
|key1|  23| 1.0|  mr|
|key1|  10| 2.0|  mr|
+----+----+----+----+