Spark API 全集(2):Spark SQL 函数全集

Posted on Edited on

Summary

org.apache.spark.sql.functions是一个Object，提供了约两百多个函数。

大部分函数与Hive的差不多。

除UDF函数，均可在spark-sql中直接使用。

经过import org.apache.spark.sql.functions._ ，也可以用于Dataframe，Dataset。

version
2.3.0

大部分支持Column的函数也支持String类型的列名。这些函数的返回类型基本都是Column。

函数很多，都在下面了。

聚合函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
approx_count_distinct
count_distinct近似值

avg
平均值

collect_list
聚合指定字段的值到list

collect_set
聚合指定字段的值到set

corr
计算两列的Pearson相关系数

count
计数

countDistinct
去重计数 SQL中用法
select count(distinct class)

covar_pop
总体协方差（population covariance）

covar_samp
样本协方差（sample covariance）

first
分组第一个元素

last
分组最后一个元素

grouping


grouping_id


kurtosis
 计算峰态(kurtosis)值

skewness
 计算偏度(skewness)

max
最大值

min
最小值

mean
平均值

stddev
 即stddev_samp

stddev_samp
 样本标准偏差（sample standard deviation）

stddev_pop
总体标准偏差（population standard deviation）

sum
求和

sumDistinct
非重复值求和 SQL中用法
select sum(distinct class)

var_pop
总体方差（population variance）

var_samp
样本无偏方差（unbiased variance）

variance
即var_samp

集合函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
array_contains(column,value)
检查array类型字段是否包含指定元素

explode
 展开array或map为多行

explode_outer
同explode，但当array或map为空或null时，会展开为null。

posexplode
同explode，带位置索引。

posexplode_outer
同explode_outer，带位置索引。

from_json
解析JSON字符串为StructType or ArrayType，有多种参数形式，详见文档。

to_json
转为json字符串，支持StructType, ArrayType of StructTypes, a MapType or ArrayType of MapTypes。

get_json_object(column,path)
获取指定json路径的json对象字符串。
select get_json_object('{"a"1,"b":2}','$.a');
[JSON Path介绍](http://blog.csdn.net/koflance/article/details/63262484)

json_tuple(column,fields)
获取json中指定字段值。select json_tuple('{"a":1,"b":2}','a','b');

map_keys
返回map的键组成的array

map_values
返回map的值组成的array

size
array or map的长度

sort_array(e: Column, asc: Boolean)
将array中元素排序（自然排序），默认asc。

时间函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
add_months(startDate: Column, numMonths: Int)
指定日期添加n月

date_add(start: Column, days: Int)
指定日期之后n天 e.g. select date_add('2018-01-01',3)

date_sub(start: Column, days: Int)
指定日期之前n天

datediff(end: Column, start: Column)
两日期间隔天数

current_date()
当前日期

current_timestamp()
当前时间戳，TimestampType类型

date_format(dateExpr: Column, format: String)
日期格式化

dayofmonth(e: Column)
日期在一月中的天数，支持 date/timestamp/string

dayofyear(e: Column)
日期在一年中的天数， 支持 date/timestamp/string

weekofyear(e: Column)
日期在一年中的周数， 支持 date/timestamp/string

from_unixtime(ut: Column, f: String)
时间戳转字符串格式

from_utc_timestamp(ts: Column, tz: String)
时间戳转指定时区时间戳

to_utc_timestamp(ts: Column, tz: String)
指定时区时间戳转UTF时间戳

hour(e: Column)
提取小时值

minute(e: Column)
提取分钟值

month(e: Column)
提取月份值

quarter(e: Column)
提取季度

second(e: Column)
提取秒

year(e: Column):提取年

last_day(e: Column)
指定日期的月末日期

months_between(date1: Column, date2: Column)
计算两日期差几个月

next_day(date: Column, dayOfWeek: String)
计算指定日期之后的下一个周一、二...，dayOfWeek区分大小写，只接受 "Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"。

to_date(e: Column)
字段类型转为DateType

trunc(date: Column, format: String)
日期截断

unix_timestamp(s: Column, p: String)
指定格式的时间字符串转时间戳

unix_timestamp(s: Column)
同上，默认格式为 yyyy-MM-dd HH:mm:ss

unix_timestamp():当前时间戳(秒),底层实现为unix_timestamp(current_timestamp(), yyyy-MM-dd HH:mm:ss)

window(timeColumn: Column, windowDuration: String, slideDuration: String, startTime: String)
时间窗口函数，将指定时间(TimestampType)划分到窗口

数学函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
cos,sin,tan
计算角度的余弦，正弦。。。

sinh,tanh,cosh
计算双曲正弦，正切，。。

acos,asin,atan,atan2
计算余弦/正弦值对应的角度

bin
将long类型转为对应二进制数值的字符串For example, bin("12") returns "1100".

bround
舍入，使用Decimal的HALF_EVEN模式，v>0.5向上舍入，v< 0.5向下舍入，v0.5向最近的偶数舍入。

round(e: Column, scale: Int)
HALF_UP模式舍入到scale为小数点。v>=0.5向上舍入，v< 0.5向下舍入,即四舍五入。

ceil
向上舍入

floor
向下舍入

cbrt
Computes the cube-root of the given value.

conv(num:Column, fromBase: Int, toBase: Int)
 转换数值（字符串）的进制

log(base: Double, a: Column):$log_{base}(a)$

log(a: Column):$log_e(a)$

log10(a: Column):$log_{10}(a)$

log2(a: Column):$log_{2}(a)$

log1p(a: Column):$log_{e}(a+1)$

pmod(dividend: Column, divisor: Column):Returns the positive value of dividend mod divisor.

pow(l: Double, r: Column):$r^l$ 注意r是列

pow(l: Column, r: Double):$r^l$ 注意l是列

pow(l: Column, r: Column):$r^l$ 注意r,l都是列

radians(e: Column):角度转弧度

rint(e: Column):Returns the double value that is closest in value to the argument and is equal to a mathematical integer.

shiftLeft(e: Column, numBits: Int):向左位移

shiftRight(e: Column, numBits: Int):向右位移

shiftRightUnsigned(e: Column, numBits: Int):向右位移（无符号位）

signum(e: Column):返回数值正负符号

sqrt(e: Column):平方根

hex(column: Column):转十六进制

unhex(column: Column):逆转十六进制

混杂(misc)函数

1
2
3
4
5
6
7
8
9
crc32(e: Column):计算CRC32,返回bigint

hash(cols: Column*):计算 hash code，返回int

md5(e: Column):计算MD5摘要，返回32位，16进制字符串

sha1(e: Column):计算SHA-1摘要，返回40位，16进制字符串

sha2(e: Column, numBits: Int):计算SHA-1摘要，返回numBits位，16进制字符串。numBits支持224, 256, 384, or 512.

其他非聚合函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
abs(e: Column)
绝对值

array(cols: Column*)
多列合并为array，cols必须为同类型

map(cols: Column*):
将多列组织为map，输入列必须为（key,value)形式，各列的key/value分别为同一类型。

bitwiseNOT(e: Column):
Computes bitwise NOT.

broadcast[T](df: Dataset[T]): Dataset[T]:
将df变量广播，用于实现broadcast join。如left.join(broadcast(right), "joinKey")

coalesce(e: Column*):
返回第一个非空值

col(colName: String):
返回colName对应的Column

column(colName: String):
col函数的别名

expr(expr: String):
解析expr表达式，将返回值存于Column，并返回这个Column。

greatest(exprs: Column*):
返回多列中的最大值，跳过Null

least(exprs: Column*):
返回多列中的最小值，跳过Null

input_file_name():返
回当前任务的文件名 ？？

isnan(e: Column):
检查是否NaN（非数值）

isnull(e: Column):
检查是否为Null

lit(literal: Any):
将字面量(literal)创建一个Column

typedLit[T](literal: T)(implicit arg0: scala.reflect.api.JavaUniverse.TypeTag[T]):
将字面量(literal)创建一个Column，literal支持 scala types e.g.: List, Seq and Map.

monotonically_increasing_id():
返回单调递增唯一ID，但不同分区的ID不连续。ID为64位整型。

nanvl(col1: Column, col2: Column):
col1为NaN则返回col2

negate(e: Column):
负数，同df.select( -df("amount") )

not(e: Column):
取反，同df.filter( !df("isActive") )

rand():
随机数[0.0, 1.0]

rand(seed: Long):
随机数[0.0, 1.0]，使用seed种子

randn():
随机数，从正态分布取

randn(seed: Long):
同上

spark_partition_id():
返回partition ID

struct(cols: Column*):
多列组合成新的struct column ？？

when(condition: Column, value: Any):
当condition为true返回value，如
people.select(when(people("gender") === "male", 0)
  .when(people("gender") === "female", 1)
  .otherwise(2))
如果没有otherwise且condition全部没命中，则返回null.

排序函数

1
2
3
4
5
6
7
8
9
10
11
asc(columnName: String):正序

asc_nulls_first(columnName: String):正序，null排最前

asc_nulls_last(columnName: String):正序，null排最后

e.g.
df.sort(asc("dept"), desc("age"))

对应有desc函数
 desc,desc_nulls_first,desc_nulls_last

字符串函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

ascii(e: Column): 计算第一个字符的ascii码

base64(e: Column): base64转码

unbase64(e: Column): base64解码

concat(exprs: Column*):连接多列字符串

concat_ws(sep: String, exprs: Column*):使用sep作为分隔符连接多列字符串

decode(value: Column, charset: String): 解码

encode(value: Column, charset: String): 转码，charset支持 'US-ASCII', 'ISO-8859-1', 'UTF-8', 'UTF-16BE', 'UTF-16LE', 'UTF-16'。

format_number(x: Column, d: Int):格式化'#,###,###.##'形式的字符串

format_string(format: String, arguments: Column*): 将arguments按format格式化，格式为printf-style。

initcap(e: Column): 单词首字母大写

lower(e: Column): 转小写

upper(e: Column): 转大写

instr(str: Column, substring: String): substring在str中第一次出现的位置

length(e: Column): 字符串长度

levenshtein(l: Column, r: Column): 计算两个字符串之间的编辑距离（Levenshtein distance）

locate(substr: String, str: Column): substring在str中第一次出现的位置，位置编号从1开始，0表示未找到。

locate(substr: String, str: Column, pos: Int): 同上，但从pos位置后查找。

lpad(str: Column, len: Int, pad: String):字符串左填充。用pad字符填充str的字符串至len长度。有对应的rpad，右填充。

ltrim(e: Column):剪掉左边的空格、空白字符，对应有rtrim.

ltrim(e: Column, trimString: String):剪掉左边的指定字符,对应有rtrim.

trim(e: Column, trimString: String):剪掉左右两边的指定字符

trim(e: Column):剪掉左右两边的空格、空白字符

regexp_extract(e: Column, exp: String, groupIdx: Int): 正则提取匹配的组

regexp_replace(e: Column, pattern: Column, replacement: Column): 正则替换匹配的部分，这里参数为列。

regexp_replace(e: Column, pattern: String, replacement: String): 正则替换匹配的部分

repeat(str: Column, n: Int):将str重复n次返回

reverse(str: Column): 将str反转

soundex(e: Column): 计算桑迪克斯代码（soundex code）PS:用于按英语发音来索引姓名,发音相同但拼写不同的单词，会映射成同一个码。

split(str: Column, pattern: String): 用pattern分割str

substring(str: Column, pos: Int, len: Int): 在str上截取从pos位置开始长度为len的子字符串。

substring_index(str: Column, delim: String, count: Int):Returns the substring from string str before count occurrences of the delimiter delim. If count is positive, everything the left of the final delimiter (counting from left) is returned. If count is negative, every to the right of the final delimiter (counting from the right) is returned. substring_index performs a case-sensitive match when searching for delim.

translate(src: Column, matchingString: String, replaceString: String):把src中的matchingString全换成replaceString。

UDF函数

1
2
3
4
5
6
7
8
9
10
11
12
user-defined function.

callUDF(udfName: String, cols: Column*): 调用UDF
import org.apache.spark.sql._

val df = Seq(("id1", 1), ("id2", 4), ("id3", 5)).toDF("id", "value")
val spark = df.sparkSession
spark.udf.register("simpleUDF", (v: Int) => v * v)
df.select($"id", callUDF("simpleUDF", $"value"))

udf: 定义UDF

窗口函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

cume_dist(): cumulative distribution of values within a window partition

currentRow(): returns the special frame boundary that represents the current row in the window partition.

rank():排名，返回数据项在分组中的排名，排名相等会在名次中留下空位 1,2,2,4。

dense_rank(): 排名，返回数据项在分组中的排名，排名相等会在名次中不会留下空位 1,2,2,3。

row_number():行号，为每条记录返回一个数字 1,2,3,4

percent_rank():returns the relative rank (i.e. percentile) of rows within a window partition.

lag(e: Column, offset: Int, defaultValue: Any): offset rows before the current row

lead(e: Column, offset: Int, defaultValue: Any): returns the value that is offset rows after the current row

ntile(n: Int): returns the ntile group id (from 1 to n inclusive) in an ordered window partition.

unboundedFollowing():returns the special frame boundary that represents the last row in the window partition.

参考