Spark-SQL 使用SQL和DSL 統計使用者上網流量案例

阿新 • • 發佈：2021-01-07

技術標籤：Spark spark

需求分析：

統計使用者上網流量，如果兩次上網的時間小於10分鐘，就可以rollup到一起

uid,start_time,end_time,flow
1,2020-02-18 14:20:30,2020-02-18 14:46:30,20
1,2020-02-18 14:47:20,2020-02-18 15:20:30,30
1,2020-02-18 15:37:23,2020-02-18 16:05:26,40
1,2020-02-18 16:06:27,2020-02-18 17:20:49,50
1,2020-02-18 17:21:50,2020-02-18 18:03:27,60
2,2020-02-18 14:18:24,2020-02-18 15:01:40,20

2,2020-02-18 15:20:49,2020-02-18 15:30:24,30
2,2020-02-18 16:01:23,2020-02-18 16:40:32,40
2,2020-02-18 16:44:56,2020-02-18 17:40:52,50
3,2020-02-18 14:39:58,2020-02-18 15:35:53,20
3,2020-02-18 15:36:39,2020-02-18 15:24:54,30

--向下推一格

select
uid,
start_time,
end_time,
flow,
lag(end_time,1,start_time) over(partition by uid order by start_time) lag_time

from
t_int

+---+-------------------+-------------------+----+-------------------+
|uid| start_time| end_time|flow| lag_time|
+---+-------------------+-------------------+----+-------------------+
| 3|2020-02-18 14:39:58|2020-02-18 15:35:53| 20| null|
| 3|2020-02-18 15:36:39|2020-02-18 15:24:54| 30|2020-02-18 15:35:53|

| 1|2020-02-18 14:20:30|2020-02-18 14:46:30| 20| null|
| 1|2020-02-18 14:47:20|2020-02-18 15:20:30| 30|2020-02-18 14:46:30|
| 1|2020-02-18 15:37:23|2020-02-18 16:05:26| 40|2020-02-18 15:20:30|
| 1|2020-02-18 16:06:27|2020-02-18 17:20:49| 50|2020-02-18 16:05:26|
| 1|2020-02-18 17:21:50|2020-02-18 18:03:27| 60|2020-02-18 17:20:49|
| 2|2020-02-18 14:18:24|2020-02-18 15:01:40| 20| null|
| 2|2020-02-18 15:20:49|2020-02-18 15:30:24| 30|2020-02-18 15:01:40|
| 2|2020-02-18 16:01:23|2020-02-18 16:40:32| 40|2020-02-18 15:30:24|
| 2|2020-02-18 16:44:56|2020-02-18 17:40:52| 50|2020-02-18 16:40:32|
+---+-------------------+-------------------+----+-------------------+

--用lag_time 減去 start_time 如果差值大於10分鐘返回1，小於10分鐘返回0

select
uid,
start_time,
end_time,
flow,
if((to_unix_timestamp(start_time)-to_unix_timestamp(lag_time))/60>10,1,0) flag
from
(
select
uid,
start_time,
end_time,
flow,
lag(end_time,1,start_time) over(partition by uid order by start_time) lag_time
from
t_int
)t1

+---+-------------------+-------------------+----+----+
|uid| start_time| end_time|flow|flag|
+---+-------------------+-------------------+----+----+
| 3|2020-02-18 14:39:58|2020-02-18 15:35:53| 20| 0|
| 3|2020-02-18 15:36:39|2020-02-18 15:24:54| 30| 0|
| 1|2020-02-18 14:20:30|2020-02-18 14:46:30| 20| 0|
| 1|2020-02-18 14:47:20|2020-02-18 15:20:30| 30| 0|
| 1|2020-02-18 15:37:23|2020-02-18 16:05:26| 40| 1|
| 1|2020-02-18 16:06:27|2020-02-18 17:20:49| 50| 0|
| 1|2020-02-18 17:21:50|2020-02-18 18:03:27| 60| 0|
| 2|2020-02-18 14:18:24|2020-02-18 15:01:40| 20| 0|
| 2|2020-02-18 15:20:49|2020-02-18 15:30:24| 30| 1|
| 2|2020-02-18 16:01:23|2020-02-18 16:40:32| 40| 1|
| 2|2020-02-18 16:44:56|2020-02-18 17:40:52| 50| 0|
+---+-------------------+-------------------+----+----+

--根據視窗函式聚合flag
select
uid,
start_time,
end_time,
flow,
sum(flag) over(partition by uid order by start_time) sum_flag
from
(
select
uid,
start_time,
end_time,
flow,
if((to_unix_timestamp(start_time)-to_unix_timestamp(lag_time))/60>10,1,0) flag
from
(
select
uid,
start_time,
end_time,
flow,
lag(end_time,1,start_time) over(partition by uid order by start_time) lag_time
from
t_int
)t1
)t2

+---+-------------------+-------------------+----+--------+
|uid| start_time| end_time|flow|sum_flag|
+---+-------------------+-------------------+----+--------+
| 3|2020-02-18 14:39:58|2020-02-18 15:35:53| 20| 0|
| 3|2020-02-18 15:36:39|2020-02-18 15:24:54| 30| 0|
| 1|2020-02-18 14:20:30|2020-02-18 14:46:30| 20| 0|
| 1|2020-02-18 14:47:20|2020-02-18 15:20:30| 30| 0|
| 1|2020-02-18 15:37:23|2020-02-18 16:05:26| 40| 1|
| 1|2020-02-18 16:06:27|2020-02-18 17:20:49| 50| 1|
| 1|2020-02-18 17:21:50|2020-02-18 18:03:27| 60| 1|
| 2|2020-02-18 14:18:24|2020-02-18 15:01:40| 20| 0|
| 2|2020-02-18 15:20:49|2020-02-18 15:30:24| 30| 1|
| 2|2020-02-18 16:01:23|2020-02-18 16:40:32| 40| 2|
| 2|2020-02-18 16:44:56|2020-02-18 17:40:52| 50| 2|
+---+-------------------+-------------------+----+--------+

--累加flow

select
uid,
min(start_time) start_time,
max(end_time) end_time,
sum(flow) sum_flow
from
(
select
uid,
start_time,
end_time,
flow,
sum(flag) over(partition by uid order by start_time) sum_flag
from
(
select
uid,
start_time,
end_time,
flow,
if((to_unix_timestamp(start_time)-to_unix_timestamp(lag_time))/60>10,1,0) flag
from
(
select
uid,
start_time,
end_time,
flow,
lag(end_time,1,start_time) over(partition by uid order by start_time) lag_time
from
t_int
)t1
)t2
)t3
group by uid,sum_flag

+---+-------------------+-------------------+--------+
|uid| start_time| end_time|sum_flow|
+---+-------------------+-------------------+--------+
| 3|2020-02-18 14:39:58|2020-02-18 15:35:53| 50.0|
| 1|2020-02-18 14:20:30|2020-02-18 15:20:30| 50.0|
| 1|2020-02-18 15:37:23|2020-02-18 18:03:27| 150.0|
| 2|2020-02-18 14:18:24|2020-02-18 15:01:40| 20.0|
| 2|2020-02-18 15:20:49|2020-02-18 15:30:24| 30.0|
| 2|2020-02-18 16:01:23|2020-02-18 17:40:52| 90.0|
+---+-------------------+-------------------+--------+

1.使用SQL

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

object SQLFlowRollupDemo {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      .getOrCreate()

    val df: DataFrame = spark.read
      .option("header", "true")
      .csv("src/main/scala/data/data.csv")

    df.createTempView("t_int")

    val res = spark.sql(
      """
        |
        |select
        |  uid,
        |  min(start_time) start_time,
        |  max(end_time) end_time,
        |  sum(flow) sum_flow
        |from
        |(
        |  select
        |    uid,
        |    start_time,
        |    end_time,
        |    flow,
        |    sum(flag) over(partition by uid order by start_time) sum_flag
        |  from
        |  (
        |    select
        |      uid,
        |      start_time,
        |      end_time,
        |      flow,
        |      if((to_unix_timestamp(start_time)-to_unix_timestamp(lag_time))/60>10,1,0) flag
        |    from
        |    (
        |      select
        |        uid,
        |        start_time,
        |        end_time,
        |        flow,
        |        lag(end_time,1,start_time) over(partition by uid order by start_time) lag_time
        |      from
        |        t_int
        |    )t1
        |  )t2
        |)t3
        |group by uid,sum_flag
        |
        |""".stripMargin)

    res.show()

    spark.stop()
  }

}

2.使用DSL

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object DSLFlowRollupDemo {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      .getOrCreate()

    val df: DataFrame = spark.read
      .option("header", "true")
      .csv("src/main/scala/data/data.csv")

    import spark.implicits._
    import org.apache.spark.sql.functions._
    val res = df.select('uid,
      'start_time,
      'end_time,
      'flow,
      expr("lag(end_time,1,start_time)")
        over(Window.partitionBy('uid) orderBy("start_time"))as "lag_time")
      .select('uid,'start_time,'end_time,'flow,
      expr("if((to_unix_timestamp(start_time)-to_unix_timestamp(lag_time))/60 > 10,1,0)") as "flag"
      )
      .select('uid,'start_time,'end_time,'flow,
        sum('flag) over(Window.partitionBy("uid") orderBy("start_time")) as "sum_flag"
      )
      .groupBy("uid","sum_flag")
      .agg('uid,
        min("start_time") as "start_time",
        max("end_time") as "end_time",
        sum('flow)
      )

    res.show()

    spark.stop()
  }
}

Spark-SQL 使用SQL和DSL 統計使用者上網流量案例

技術標籤：Sparkspark 需求分析：統計使用者上網流量，如果兩次上網的時間小於10分鐘，就可以rollup到一起

Spark解決SQL和RDDjoin結果不一致問題（工作實錄）

問題描述：DataFrame的join結果不正確，dataframeA(6000無重複條資料) join dataframeB(220條無重複資料，由dataframeA轉化而來，key值均源於dataframeA) 只有200條資料，丟了20條

關於SQL語句轉DSL語句的一些實踐和思考(一)

現在接觸的專案是公司的路由中臺,每天的資料量是億級別的,同時要記錄每一次請求的詳細資料

閒來無事瞭解下資料庫 - SQL概述和資料定義 - 1

概覽 SQL概述什麼是SQL? (Structured Query Language,SQL) SQL的全稱就是結構化查詢語言,是專門用來與資料庫通訊的語言,它可以幫助使用者操作關係資料庫,包括新增,刪除,修改和查詢資料,定義和修資料模式等. 相當於

SELECT INTO 和 INSERT INTO SELECT 兩種表複製語句詳解(SQL資料庫和Oracle資料庫的區別)

1.INSERT INTO SELECT語句語句形式為：Insert into Table2(field1,field2,...) select value1,value2,... from Table1

MySQL基礎-SQL命令和語言

目錄一.MySQL命令1.mysql中的快捷鍵2.mysql的help命令3.客戶端mysqladmin命令二.SQL語句三.DDL:資料定義語言1.create針對庫的操作1).語法2).建立庫3).檢視建庫語句4).建立庫並指定字符集5).修改庫(字符集)2.create針

mysql 慢查詢慢SQL 記錄和調整

在mysql 中，可以通過設定配置引數，開啟慢 SQL 的記錄在 my.cnf 的 [mysqld] 配置下，可以設定以下引數實現慢查詢記錄

sql join 和 left join 區別 on 和 where 區別

有三張表film category film_category left join 查詢結果 select *from film f left join film_category fc on f.film_id=fc.f_id

分享一個php的防火牆，攔截SQL注入和xss

一個基於php的防火牆程式，攔截sql注入和xss攻擊等安裝 composer require xielei/waf 使用說明

SQL Server 2008下輕鬆除錯T-SQL語句和儲存過程

一、除錯T-SQL語句： 1.Debug普通T-SQL語句: SQL程式碼如下： 1 use northwind 2 go 3 declare @i int ,@j int,@k int

MyBatis-動態SQL使用和原理

參考： https://www.cnblogs.com/ysocean/p/7289529.html https://www.cnblogs.com/fangjian0423/p/mybaits-dynamic-sql-analysis.html

開源的SQL編輯和資料庫管理器

Beekeeper Studio是一款開源的SQL編輯器和資料庫管理器，可以支援MySQL，Postgres的，SQLite，SQL等伺服器。Beekeeper Studio可以使用SSL加密連線，或者通過SSH建立隧道，讓大家的資料更加安全。如果大家需要一款開

演算法-一個經典sql 題和一個Java演算法題

1.sql題描述話說有一個日誌表，只有兩列，分別是連續id和num 至於啥意思，把它當金額把。現在想知道連續次數3次及以上的num，資料如下

資料庫實驗五 SQL語句和E-R模型

實驗目的掌握複雜SQL語句掌握E-R模型實驗內容 1、針對以下關係模式: E企業（企業編號，企業名）

一文為你詳解Unique SQL原理和應用

摘要：以一定的演算法結合解析樹中的各結點，計算出來一個整數值，用來唯一標識這一類SQL，這個整數值被稱為Unique SQL ID，Unique SQL ID相同的SQL語句屬於同一個“Unique SQL”。

SQL update和alter之間的區別

alter用來修改基本表,是對錶的結構進行操作，在已有的表中新增、刪除或修改列

在postgreSQL中執行sql指令碼和pg_restore命令方式

今天踩坑了，把powerdesign生成的sql指令碼檔案，用pg_restore命令一直執行。。。

SQL UNION 和 UNION ALL 操作符

技術標籤：MYSQL UNION 操作符用於合併兩個或多個 SELECT 語句的,除去重複的結果集。

BeanUtils中DateConverter does not support default String to ‘Date‘ 以及使用sql.date和util.date區別

技術標籤：JAVA筆記javasqlbeanutilsmysql資料庫 BeanUtils BeanUtils是Apache commens元件裡面的成員，由Apache提供的一套開源api，用於簡化對javaBean的操作，能夠對基本型別自動轉換。

[網摘]SQL GUID和自增列做主鍵的優缺點

SQL GUID和自增列做主鍵的優缺點公司的資料庫全部是使用GUID做主鍵的，很多人習慣使用int做主鍵。所以呢，這裡總結一下，將兩種資料型別做主鍵進行一個比較。

Spark-SQL 使用SQL和DSL 統計使用者上網流量 案例

1.使用SQL

2.使用DSL

相關推薦

Spark-SQL 使用SQL和DSL 統計使用者上網流量案例