2. 程式人生 > >Spark SQL基本操作以及函式的使用

Spark SQL基本操作以及函式的使用

阿新 • • 發佈：2018-12-02

引語：

  本篇部落格主要介紹了Spark SQL中的filter過濾資料、去重、集合等基本操作，以及一些常用日期函式，隨機函式，字串操作等函式的使用，並列編寫了示例程式碼，同時還給出了程式碼當中用到的一些資料，放在最文章最後。

SparkSQL簡介

  Spark SQL是Spark生態系統中非常重要的元件，其前身為Shark。Shark是Spark上的資料倉庫，最初設計成與Hive相容，但是該專案於2014年開始停止開發，轉向Spark SQL。Spark SQL全面繼承了Shark，並進行了優化。 Spark SQL增加了SchemaRDD（即帶有Schema資訊的RDD），使使用者可以在Spark SQL中執行SQL語句，資料既可以來自RDD，也可以來自Hive、HDFS、Cassandra等外部資料來源，還可以是JSON格式的資料。Spark SQL目前支援Scala、Java、Python三種語言，支援SQL-92規範。

Spark SQL的優點

  Spark SQL可以很好地支援SQL查詢，一方面，可以編寫Spark應用程式使用SQL語句進行資料查詢，另一方面，也可以使用標準的資料庫聯結器（比如JDBC或ODBC）連線Spark進行SQL查詢 。

Spark SQL基本操作

去重

  distinct：根據每條資料進行完整去重。

  dropDuplicates：根據欄位去重。

  package spark2x

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

/**
  * 類名  DistinctDemo
  * 作者   彭三青
  * 建立時間  2018-11-29 15:02
  * 版本  1.0
  * 描述: $ 去重操作:distinct、drop
  */

object DistinctDemo {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[2]")
      .appName("Operations")
      .getOrCreate()
    import spark.implicits._

    val employeeDF: DataFrame = spark.read.json("E://temp/person.json")
    val employeeDS: Dataset[Employee] = employeeDF.as[Employee]

    println("--------------------distinct---------------------")
	// 根據每條資料進行完整的去重
    employeeDS.distinct().show()

    println("--------------------dropDuplicates---------------------")
    // 根據欄位進行去重
    employeeDS.dropDuplicates(Seq("name")).show()
  }
}
case class Employee(name: String, age: Long, depId: Long, gender: String, salary: Double)
過濾
filter()：括號裡的引數可以是過濾函式、函式返回的Boolean值（為true則保留，false則過濾掉）、列名或者表示式。

except：過濾出當前DataSet中有，但在另一個DataSet中不存在的。

intersect：獲取兩個DataSet的交集。

  提示：except和intersect使用的時候必須要是相同的例項，如果把另外一個的Employee換成一個同樣的欄位的Person類就會報錯。

package spark2x

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

/**
  * 類名  FilterDemo
  * 作者   彭三青
  * 建立時間  2018-11-29 15:09
  * 版本  1.0
  * 描述: $
  */

object FilterDemo {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[2]")
      .appName("FilterDemo")
      .getOrCreate()
    import spark.implicits._

    val employeeDF: DataFrame = spark.read.json("E://temp/employee.json")
    val employeeDS: Dataset[Employee] = employeeDF.as[Employee]
    val employee2DF: DataFrame = spark.read.json("E://temp/employee2.json")
    val employee2DS: Dataset[Employee] = employee2DF.as[Employee]

    println("--------------------employee--------------------")
    employeeDS.show()

    println("--------------------employee2--------------------")
    employee2DS.show()

    println(
      "       ┏┓　　　┏┓\n" +
      "     ┏┛┻━━━┛┻┓\n" +
      " 　　┃　　　　　　　┃\n" +
      " 　　┃　　　━　　　┃\n" +
      " 　　┃　┳┛　┗┳　┃\n" +
      " 　　┃　　　　　　　┃\n" +
      " 　　┃　　　┻　　　┃\n" +
      " 　　┃　　　　　　　┃\n" +
      " 　　┗━┓　　　┏━┛\n" +
      " 　　　　┃　　　┃\n" +
      " 　　　　 ┃　　　┃\n" +
      " 　　　　┃　　　┗━━━┓\n" +
      " 　　　　┃　　　　　　　┣┓\n" +
      " 　　　　┃　　　　　　　┏┛\n" +
      " 　　　　┗┓┓┏━┳┓┏┛\n" +
      " 　　　　　┃┫┫　┃┫┫\n" +
      " 　　　　　┗┻┛　┗┻┛\n"
    )

    println("-------------------------------------------------")

    // 如果引數返回true，就保留該元素，否則就過濾掉
    employeeDS.filter(employee => employee.age == 35).show()
    employeeDS.filter(employee => employee.age > 30).show()
    // 獲取當前的DataSet中有，但是在另外一個DataSet中沒有的元素
    employeeDS.except(employee2DS).show()
    // 獲取兩個DataSet的交集
    employeeDS.intersect(employee2DS).show()

    spark.stop()
  }
}
case class Employee(name: String, age: Long, depId: Long, gender: String, salary: Double)
集合
collect_set：將一個分組內指定欄位的值都收集到一起，不去重

collect_list：講一個分組內指定欄位的值都收集到一起，會去重

package spark2x

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

/**
  * 類名  CollectSetAndList
  * 作者   彭三青
  * 建立時間  2018-11-29 15:24
  * 版本  1.0
  * 描述: $ collect_list、 collect_set
  */

object CollectSetAndList {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[2]")
      .appName("FilterDemo")
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    val employeeDF: DataFrame = spark.read.json("E://temp/employee.json")
    val employeeDS: Dataset[Employee] = employeeDF.as[Employee]

    // collect_list：將一個分組內指定欄位的值都收集到一起，不去重
    // collect_set：同上，但唯一區別是會去重
    employeeDS
      .groupBy(employeeDS("depId"))
      .agg(collect_set(employeeDS("name")), collect_list(employeeDS("name")))
      .show()
  }
}
case class Employee(name: String, age: Long, depId: Long, gender: String, salary: Double)
join和sort
package spark2x

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

/**
  * 類名  JoinAndSort
  * 作者   彭三青
  * 建立時間  2018-11-29 15:19
  * 版本  1.0
  * 描述: $
  */

object JoinAndSort {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[2]")
      .appName("FilterDemo")
      .getOrCreate()
    import spark.implicits._

    val employeeDF: DataFrame = spark.read.json("E://temp/employee.json")
    val employeeDS: Dataset[Employee] = employeeDF.as[Employee]
    val departmentDF: DataFrame = spark.read.json("E://temp/department.json")
    val departmentDS: Dataset[Department] = departmentDF.as[Department]

    println("----------------------employeeDS----------------------")
    employeeDS.show()
    println("----------------------departmentDS----------------------")
    departmentDS.show()
    println("------------------------------------------------------------")

    // 等值連線
    employeeDS.joinWith(departmentDS, $"depId" === $"id").show()
    // 按照年齡進行排序，並降序排列
    employeeDS.sort($"age".desc).show()
  }
}
case class Department(id: Long, name: String)
case class Employee(name: String, age: Long, depId: Long, gender: String, salary: Double)

函式的使用

日期函式

  current_time()：獲取當前日期。

  current_timestamp()：獲取當前時間戳。

數學函式

  rand()：生成0~1之間的隨機數

  round(e: column，scale: Int )：column列名，scala精確到小數點的位數。

  round(e: column)：一個引數預設精確到小數點1位。

字串函式

  concat_ws(seq: String, exprs: column*)：字串拼接。引數seq傳入的拼接的字元，column傳入的需要拼接的字元，可以指定多個列，不同列之間用逗號隔開。

package spark2x

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

/**
  * 類名  FunctionsDemo
  * 作者   彭三青
  * 建立時間  2018-11-29 15:56
  * 版本  1.0
  * 描述: $
  */

object FunctionsDemo {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[2]")
      .appName("Operations")
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    val employeeDF: DataFrame = spark.read.json("E://temp/employee.json")
    val employeeDS: Dataset[Employee] = employeeDF.as[Employee]

    employeeDS
        .select(employeeDS("name"), current_date(), current_timestamp(),
          rand(), round(employeeDS("salary"), 2),// 取隨機數,
          concat(employeeDS("gender"), employeeDS("age")),
          concat_ws("|", employeeDS("gender"), employeeDS("age"))).show()

    spark.stop()
  }
}
case class Employee(name: String, age: Long, depId: Long, gender: String, salary: Double)
資料

  employee.json

{"name": "Leo", "age": 25, "depId": 1, "gender": "male", "salary": 20000.123}
{"name": "Marry", "age": 30, "depId": 2, "gender": "female", "salary": 25000}
{"name": "Jack", "age": 35, "depId": 1, "gender": "male", "salary": 15000}
{"name": "Tom", "age": 42, "depId": 3, "gender": "male", "salary": 18000}
{"name": "Kattie", "age": 21, "depId": 3, "gender": "female", "salary": 21000}
{"name": "Jen", "age": 30, "depId": 2, "gender": "female", "salary": 28000}
{"name": "Jen", "age": 19, "depId": 2, "gender": "male", "salary": 8000}
{"name": "Tom", "age": 42, "depId": 3, "gender": "male", "salary": 18000}
{"name": "XiaoFang", "age": 18, "depId": 3, "gender": "female", "salary": 58000}

  employee2.json

{"name": "Leo", "age": 25, "depId": 1, "gender": "male", "salary": 20000.123}
{"name": "Marry", "age": 30, "depId": 2, "gender": "female", "salary": 25000}
{"name": "Jack", "age": 35, "depId": 1, "gender": "male", "salary": 15000}
{"name": "Tom", "age": 42, "depId": 3, "gender": "male", "salary": 18000}
{"name": "Kattie", "age": 21, "depId": 3, "gender": "female", "salary": 21000}
{"name": "Jen", "age": 30, "depId": 2, "gender": "female", "salary": 28000}

  department.json

{"id": 1, "name": "Technical Department"}
{"id": 2, "name": "Financial Department"}
{"id": 3, "name": "HR Department"}

相關推薦

Spark SQL基本操作以及函式的使用

引語：   本篇部落格主要介紹了Spark SQL中的filter過濾資料、去重、集合等基本操作，以及一些常用日期函式，隨機函式，字串操作等函式的使用，並列編寫了示例程式碼，同時還給出了程式碼當中用到的一些資料，放在最文章最後。 SparkSQL簡介   Spark SQL是Sp

SQL基本操作——DROP撤銷索引、表以及數據庫

drop 操作 span run color 基本操作 strong 我們 數據 DROP撤銷索引、表以及數據庫 --DROP INDEX 命令刪除表格中的索引 DROP INDEX table_name.index_name --DROP TABLE 語句刪除表（表的結

man函式的用法，vim下的基本操作以及使用者管理

##man函式的使用## man ->>manual ###man是手冊縮寫 man man #####檢視man命令的幫助 man passwd ###檢視passwd命令的幫助 man的級別 1 系統命令 2 系統介面 3 函式庫 4

Python中集合(set)的基本操作以及一些常見的用法

python set 集合 集合的一些常見用法 Python除了List、Tuple、Dict等常用數據類型外，還有一種數據類型叫做集合（set），集合的最大特點是：集合裏邊的元素是不可重復的並且集合內的元素還是無序的，所以一般情況下集合常用的兩個場景是：1.去重（如：列表去重）；2.關

SQL基本操作

har creat mssq 基本操作 upd lena eight 列名 sqlserver 通過代碼方式創建數據庫 1 --通過代碼的方式創建數據庫 2 create database MyDatabaseNew 3 on primary 4 ( 5

SQL基本操作——ALTER

car ast body name hang cit don sch 基本 ALTER TABLE 語句用於在已有的表中添加、修改或刪除列。 Persons 表: ID LastName FirstName Address City 1 Adams John Ox

SQL基本操作——GROUP BY

order 實例 擁有 分組 tom price 金額 根據 操作 GROUP BY 語句用於結合合計函數，根據一個或多個列對結果集進行分組。 SQL GROUP BY 實例：我們擁有下面這個 "Orders" 表 O_Id OrderDate OrderPrice

SQL基本操作——日期函數

rom 兩個 server tro ima sql efault from fault SQL日期：當我們處理日期時，最難的任務恐怕是確保所插入的日期的格式，與數據庫中日期列的格式相匹配。只要數據包含的只是日期部分，運行查詢就不會出問題。但是，如果涉及時間，情況就有點復雜了

SQL基本操作——case end

兩個 span class code when font eight 成績 mage case end進行多條件的判斷 --查看Person表 select * from Person --對math字段進行條件判斷 select name,數學成績= case

SQL基本操作——declare if lese while

sele ron begin pre print str declare 輸出 col declare --第一種 declare @i int set @i= (select COUNT(*) from t8) select @i --第二種 declare @i in

SQL基本操作——select into與臨時表

weight sql基本操作 bsp 臨時表 arc table rst get 需要 SELECT INTO 語句從一個表中選取數據，然後把數據插入另一個表中，常用於創建表的備份復件或者用於對記錄進行存檔。 --制作 "Persons" 表的備份復件： SELECT *

SQL基本操作——事務

內部 隔離性 成功 應該 包括 不能 隔離 失敗 基本 事務是並發和恢復控制的基本單元。 事務四個屬性：原子性、一致性、隔離性、持久性。 原子性：一個事務是一個不可分割的單位，事務中包括的諸多操作要麽成功要麽都失敗。 一致性：事務必須使數據庫從一個一致性狀態轉變到另一個一致

Linux下對MySQL/MariaDB數據庫的基本操作以及linux mysql添加用戶，刪除用戶，以及用戶權限的授予

信息 查看 let quit mar 普通用戶 表名 mys xxxx 文章引用地址:https://www.cnblogs.com/Glory-D/p/7518541.html、https://www.cnblogs.com/zhchoutai/p/6929103.htm

Redis客戶端基本操作以及檢視慢查詢

1.連線 redis-cli.exe -h 127.0.0.1 -p 6379 2.驗證密碼 λ redis-cli.exe -h 127.0.0.1 -p 6379127.0.0.1:6379> auth 密碼 3. redis key值獲取    

ElasticSearch實戰二（es基本操作以及IK分詞器的安裝）

1 基本概念 1.1 Node 與 Cluster Elastic 本質上是一個分散式資料庫，允許多臺伺服器協同工作，每臺伺服器可以執行多個 Elastic 例項。 單個 Elastic 例項稱為一個節點（node）。一組節點構成一個叢集（cluster）。 1.2 Index El

C語言-二叉樹基本操作以及二叉搜尋樹基本操作

功能 二叉樹操作: 建立二叉樹 遍歷二叉樹(前序,中序,後續) 計算高度 計算結點數目 清空二叉樹 空樹判斷 二叉搜尋樹操作: 插入 最值(最大值,最小值) 刪除 程式碼 #include &l

Mysql sql基本操作

一、建立資料庫，編碼格式為utf-8 　　create database s12day9 charset utf8; 二、建立表 　　use s12day9; 　　create table students( 　　　　　　　　　　　　　　id int not null auto_increment

（三）mysql SQL 基本操作

MySQL伺服器物件 mysql 伺服器物件內部分成了 4 層： 系統（DBMS）---->資料庫（DB）---->表（Table）---->欄位（Field） mysql 的基本操作 基本操作：CRUD ； 將SQL的基本操作根據操作物件

MySql：SQL常用操作、函式、事物和索引

MySQL是一個關係型資料庫管理系統，在開始學習MySQL資料庫前，讓我們先了解下RDBMS的一些術語： 資料庫: 資料庫是一些關聯表的集合。 資料表: 表是資料的矩陣，在一個數據庫中的表看起來像一個簡單的電子表格。 列：一列(資料元素) 包含了相同的資料，例如郵政編碼

C# Linq基本操作以及into、let關鍵字

概念：“查詢”也是一種指令，使用這組指令可以從給定的一個或多個數據源中檢索資料，並以指令的形式返回結果。Linq也是一種查詢，可以為程式語言提供強大的方便的查詢功能，並與其合為一體。 Linq的強大體現在它的簡潔 方便的程式碼量。比如要篩選出list中一個屬性的元素，相對於list就要遍歷判斷；