Spark Sql OLAP 高阶分析函数总结
2020-04-03 本文已影响0人
郭彦超
我们经常困惑在数据挖掘和报表分析场景中sql不会写，或者因为sql太长以至于可读性降低; 今天我为大家总结了一些Spark SQL中的高阶函数，它们将会对你的业务形成助力，百倍提升你的工作效率
GROUPING__ID,CUBE,ROLLUP
可快速实现多维度自由组合分析查询，主要应用于OLAP钻取分析场景，比如，分小时、天、月的UV数。
-
cube
cube函数 多用来实现钻取查询
将一个group by中单一维度分组后进行聚合，等价于将不同维度的GROUP BY结果集进行UNION ALL
select * from eqs_1234;
month day GROUPING__ID
------------------------------------
2015-03 2015-03-10 cookie1
2015-03 2015-03-10 cookie5
2015-03 2015-03-12 cookie7
2015-04 2015-04-12 cookie3
2015-04 2015-04-13 cookie2
2015-04 2015-04-13 cookie4
2015-04 2015-04-16 cookie4
2015-03 2015-03-10 cookie2
2015-03 2015-03-10 cookie3
2015-04 2015-04-12 cookie5
2015-04 2015-04-13 cookie6
2015-04 2015-04-15 cookie3
2015-04 2015-04-15 cookie2
2015-04 2015-04-16 cookie1
SELECT
month,
day,
COUNT(DISTINCT cookieid) AS uv
FROM eqs_1234
GROUP BY cube (month,day,(month,day))
month day uv GROUPING__ID
------------------------------------------------
2015-03 NULL 5 1
2015-04 NULL 6 1
NULL 2015-03-10 4 2
NULL 2015-03-12 1 2
NULL 2015-04-12 2 2
NULL 2015-04-13 3 2
NULL 2015-04-15 2 2
NULL 2015-04-16 2 2
2015-03 2015-03-10 4 3
2015-03 2015-03-12 1 3
2015-04 2015-04-12 2 3
2015-04 2015-04-13 3 3
2015-04 2015-04-15 2 3
2015-04 2015-04-16 2 3
## 如果不知道cube函数，那么可能会用下面的方式来实现，SQL的可读性和性能大大降低
SELECT month,NULL,COUNT(DISTINCT cookieid) AS uv,1 AS GROUPING__ID FROM lxw1234 GROUP BY month
UNION ALL
SELECT NULL,day,COUNT(DISTINCT cookieid) AS uv,2 AS GROUPING__ID FROM lxw1234 GROUP BY day
UNION ALL
SELECT month,day,COUNT(DISTINCT cookieid) AS uv,3 AS GROUPING__ID FROM lxw1234 GROUP BY month,day
-
grouping_id
标记出属于哪一类维度组合，相同的组合方式grouping_id的结果一样
SELECT name, grouping_id(), sum(age), avg(height) FROM VALUES (2, 'Alice', 165), (5, 'Bob', 180) people(age, name, height) GROUP BY cube(name, height);
NULL 2 2 165.0
Alice 0 2 165.0
NULL 2 5 180.0
NULL 3 7 172.5
Bob 0 5 180.0
Bob 1 5 180.0
Alice 1 2 165.0
-
rollup
以左侧维度为主聚合维度进行层级聚合，所有维度都为NULL时代表全部数据，rollup是cube的子集；可以快速实现由左及右的下钻分析。
SELECT name, age, count(*) FROM VALUES (2, 'Alice'), (5, 'Bob') people(age, name) GROUP BY rollup(name, age);
NULL NULL 2
Alice 2 1
Bob 5 1
Bob NULL 1
Alice NULL 1
LAG,LEAD,FIRST_VALUE,LAST_VALUE
快速获取窗口内往上或往下第几行的数据
-
lag
向上取数；lag(col,n,DEFAULT) 用于统计窗口内往上第n行值
第一个参数为列名，第二个参数为往上第n行（可选，默认为1），第三个参数为默认值（当往上第n行为NULL时候，取默认值，如不指定，则为NULL）
sql("SELECT name, age, lag(age,1) over(partition by name order by age) lag FROM VALUES (2, 'Alice'), (5, 'Bob'), (12, 'Alice') people(age, name)").show
+-----+---+----+
| name|age| lag|
+-----+---+----+
| Bob| 5|null|
|Alice| 2|null|
|Alice| 12| 2|
+-----+---+----+
-
lead
向下取数；lead(col,n,DEFAULT) 用于统计窗口内往下第n行值
第一个参数为列名，第二个参数为往下第n行（可选，默认为1），第三个参数为默认值（当往下第n行为NULL时候，取默认值，如不指定，则为NULL）
sql("SELECT name, age, lead(age,1,age) over(partition by name order by age) leadFROM VALUES (2, 'Alice') 'Bob'),, (5, 'Bob'), (12, 'Alice') people(age, name)").show
+-----+---+---+
| name|age|lead|
+-----+---+---+
| Bob| 5| 5|
|Alice| 2| 12|
|Alice| 12| 12|
+-----+---+---+
-
first_value/first
first_value(expr[, isIgnoreNull]) 用来获取分组内第一个满足expr表达式的值
第一个参数是列名 支持表达式，第二个参数可选，当为true时 只返回非NULL数值
sql("SELECT name, age, first(age,true) over(partition by name order by age) first FROM VALUES (2, 'Alice'), (5, 'Bob'), (12, 'Alice') people(age, name)").show
+-----+---+---+
| name|age|first|
+-----+---+---+
| Bob| 5| 5|
|Alice| 2| 2|
|Alice| 12| 2|
+-----+---+---+
-
last_value/last
last_value(expr[, isIgnoreNull]) 用来获取分组内最后一个满足expr表达式的值
第一个参数是列名 支持表达式，第二个参数可选，当为true时 只返回非NULL数值
sql("SELECT name, age, last(age,true) over(partition by name) last FROM VALUES (2, 'Alice'), (5, 'Bob'), (12, 'Alice'), (1, 'Alice') people(age, name)").show
+-----+---+-----+
| name|age|last|
+-----+---+-----+
| Bob| 5| 5|
|Alice| 12| 2|
|Alice| 1| 2|
|Alice| 2| 2|
+-----+---+-----+
## 如果要获取窗口排序后的末尾值，需要使用first函数实现
sql("SELECT name, age, last(age,true) over(partition by name) last1, first(age) over(partition by name order by age desc) last2 FROM VALUES (2, 'Alice'), (5, 'Bob'), (12, 'Alice'), (1, 'Alice') people(age, name)").show
+-----+---+-----+-----+
| name|age|last1|last2|
+-----+---+-----+-----+
| Bob| 5| 5| 5|
|Alice| 12| 1| 12|
|Alice| 2| 1| 12|
|Alice| 1| 1| 12|
+-----+---+-----+-----+
NTILE,ROW_NUMBER,DENSE_RANK
常用窗口函数
-
ntile
对数据按照某一维度进行等比切片，如果数据不均匀，会优先补充上面分片的数据量
sql("select name,age,ntile(2) over(partition by name order by age) ntile from VALUES (2, 'Alice'), (5, 'Bob'), (12, 'Alice'), (1, 'Alice') people(age, name)").show
+-----+---+-----+
| name|age|ntile|
+-----+---+-----+
| Bob| 5| 1|
|Alice| 1| 1|
|Alice| 2| 1|
|Alice| 12| 2|
+-----+---+-----+
sql("select name,age,ntile(3) over(partition by name order by age) ntile from VALUES (2, 'Alice'), (5, 'Bob'), (12, 'Alice'), (1, 'Alice') people(age, name)").show
+-----+---+-----+
| name|age|ntile|
+-----+---+-----+
| Bob| 5| 1|
|Alice| 1| 1|
|Alice| 2| 2|
|Alice| 12| 3|
+-----+---+-----+
## 如统计某个用户一天内pv最多的前1/3是那几天
## 那么将数据3等分后 取rn=1就是我们需要的记录
cookieid day pv rn
----------------------------------
cookie1 2015-04-12 7 1
cookie1 2015-04-11 5 1
cookie1 2015-04-15 4 1
cookie1 2015-04-16 4 2
cookie1 2015-04-13 3 2
cookie1 2015-04-14 2 3
cookie1 2015-04-10 1 3
cookie2 2015-04-15 9 1
cookie2 2015-04-16 7 1
cookie2 2015-04-13 6 1
cookie2 2015-04-12 5 2
cookie2 2015-04-14 3 2
cookie2 2015-04-11 3 3
cookie2 2015-04-10 2 3
-
row_number
row_number从1开始，按照指定字段排序后生成递增序列号
如 按照日期分组后对作品pv进行降序排列，生成组内pv排序的名次
sql("select day, sid, pv ,row_number() over(partition by day order by pv desc)rn from VALUES('2020-04-04','a1',11), ('2020-04-01','b1',51), ('2020-04-04','b1',11), ('2020-04-01','c1',21), ('2020-04-01','a1',1) log(day, sid, pv)").show
+----------+---+---+---+
| day|sid| pv| rn|
+----------+---+---+---+
|2020-04-01| b1| 51| 1|
|2020-04-01| c1| 21| 2|
|2020-04-01| a1| 1| 3|
|2020-04-04| a1| 11| 1|
|2020-04-04| b1| 11| 2|
+----------+---+---+---+
-
dense_rank
与row_number不同的是相同 pv 的序列号，dense_rank返回值是相同的
sql("select day, sid, pv, dense_rank() over(partition by day order by pv desc) dense_rank,row_number() over(partition by day order by pv desc)rn from VALUES('2020-04-04','a1',11), ('2020-04-01','b1',51), ('2020-04-04','b1',11), ('2020-04-01','c1',21), ('2020-04-01','a1',1) log(day, sid, pv)").show
+----------+---+---+----------+---+
| day|sid| pv|dense_rank| rn|
+----------+---+---+----------+---+
|2020-04-01| b1| 51| 1| 1|
|2020-04-01| c1| 21| 2| 2|
|2020-04-01| a1| 1| 3| 3|
|2020-04-04| a1| 11| 1| 1|
|2020-04-04| b1| 11| 1| 2|
+----------+---+---+----------+---+
最后补充SUM,AVG,MIN,MAX聚合函数的窗口化支持
- 统计某个作品随时间增长的累计pv
sql("select day, sid, pv, sum(pv) over(partition by sid order by day) pv1 from VALUES('2020-04-04','a1',11), ('2020-04-03','d1',51), ('2020-04-02','d1',11), ('2020-04-01','d1',21), ('2020-04-04','d1',1) log(day, sid, pv)").show
+----------+---+---+---+
| day|sid| pv|pv1|
+----------+---+---+---+
|2020-04-01| d1| 21| 21|
|2020-04-02| d1| 11| 32|
|2020-04-03| d1| 51| 83|
|2020-04-04| d1| 1| 84|
|2020-04-04| a1| 11| 11|
+----------+---+---+---+
- ROWS BETWEEN是窗口子函数，借助该函数可限定累计的范围
## ROWS BETWEEN 2 PRECEDING AND CURRENT ROW 意思是当前行pv + 往前2行pv值
sql("select day, sid, pv, sum(pv) over(partition by sid order by day) pv1, sum(pv) over(partition by sid order by day ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) pv2 from VALUES('2020-04-04','a1',11), ('2020-04-03','d1',51), ('2020-04-02','d1',11), ('2020-04-01','d1',21), ('2020-04-04','d1',1) log(day, sid, pv)").show
+----------+---+---+---+---+
| day|sid| pv|pv1|pv2|
+----------+---+---+---+---+
|2020-04-01| d1| 21| 21| 21|
|2020-04-02| d1| 11| 32| 32|
|2020-04-03| d1| 51| 83| 83|
|2020-04-04| d1| 1| 84| 63|
|2020-04-04| a1| 11| 11| 11|
+----------+---+---+---+---+
