SparkSQL函数定义——UDF函数，窗口函数

1 定义UDF函数

目前python仅支持UDF

两种定义方式：

sparksession.udf.register()

注册的UDF可以用于DSL和SQL

返回值用于DSL风格，传参内给的名字用于SQL风格
    方法一语法：
udf对象 = sparksession.udf.register( 参数1，参数2，参数3)

参数1：UDF名称，可用于SQL风格

参数2：被注册成UDF的方法名

参数3：声明UDF的返回值类型

udf 对象：返回值对象，是一个 UDF对象，可用于DSL风格

2.pyspark.sql.functions.udf

仅能用于DSL风格
    方式二语法：
udf对象 = F.udf( 参数1，参数2 )

参数1：被注册成UDF的方法名

参数2：声明UDF的返回值类型

udf 对象：返回值对象，是一个 UDF对象，可用于DSL风格

其中F是：

from pyspark.sql import functions as F

其中，被注册成UDF的方法名是指具体的计算方法，如：

def add( x, y ):x+y

add就是将要被注册成UDF的方法名

# coding:utf8
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType
from pyspark.sql import functions as F
if __name__ == '__main__':
    # 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.appName("test").master("local[*]").getOrCreate()
    sc = spark.sparkContext

    # 构建一个RDD
    rdd = sc.parallelize([1, 2, 3, 4, 5, 6, 7]).map(lambda x: [x])
    df = rdd.toDF(["num"])

    def num_ride_10(num):
        return num * 10

    # 参数1： 注册的UDF名称， 这个udf名称，仅可以用于SQL风格
    # 参数2： UDF处理逻辑，是一个单独的方法
    # 参数3： UDF的返回值类型，注意，UDF注册时候，必须声明返回值类型，并且UDF的真实返回值一定要和声明返回值一样
    # 返回值对象： 这是一个UDF对象，仅可以用于DSL语法
    # 当前这种方式定义的UDF，可以通过参数1的名称用于SQL风格，通过返回值对象用的DSL风格
    udf2 = spark.udf.register('udf1', num_ride_10, IntegerType())

    # SQL风格中使用
    # selectExpr 以SELECT的表达式执行， 表达式SQL风格的表达式(字符串)
    # select方法， 接受普通的字符串段名，或者返回值是Column 对象的计算
    df.selectExpr("udf1(num)").show()

    # DSL风格中使用
    # 返回值UDF对象， 如果作为方法使用，传入的参数一定是 Column对象
    df.select(udf2(df['num'])). show()

    # 方式2注册， 仅能用于DSL风格
    udf_3 = F.udf(num_ride_10, IntegerType())
    df.select(udf_3(df['num'])).show()

# 输出
+---------+
|udf1(num)|
+---------+
|       10|
|       20|
|       30|
|       40|
|       50|
|       60|
|       70|
+---------+

+---------+
|udf1(num)|
+---------+
|       10|
|       20|
|       30|
|       40|
|       50|
|       60|
|       70|
+---------+

+----------------+
|num_ride_10(num)|
+----------------+
|              10|
|              20|
|              30|
|              40|
|              50|
|              60|
|              70|
+----------------+

1.1 返回值是数组类型的UDF定义

# coding:utf8
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType, ArrayType
from pyspark.sql import functions as F
if __name__ == '__main__':
    # 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.appName("test").master("local[*]").getOrCreate()
    sc = spark.sparkContext

    # 构建一个RDD
    rdd = sc.parallelize([["hadoop spark flink"], ["hadoop flink java"]])
    df = rdd.toDF(["line"])

    # 注册 UDF
    def split_line(data):
        return data.split(" ")  # 返回值是一个数组对象

    # 方式1 构建UDF
    udf1 = spark.udf.register('udf1', split_line, ArrayType(StringType()))
    # DSL风格
    df.select(udf1(df["line"])).show()

    # SQL风格
    df.createTempView("line")
    spark.sql("SELECT udf1(line) FROM line").show(truncate=False)

    # 方式2构建
    udf3 = F.udf(split_line, ArrayType(StringType()))
    df.select(udf3(df["line"])).show()

# 输出
+--------------------+
|          udf1(line)|
+--------------------+
|[hadoop, spark, f...|
|[hadoop, flink, j...|
+--------------------+

+----------------------+
|udf1(line)            |
+----------------------+
|[hadoop, spark, flink]|
|[hadoop, flink, java] |
+----------------------+

+--------------------+
|    split_line(line)|
+--------------------+
|[hadoop, spark, f...|
|[hadoop, flink, j...|
+--------------------+

1.2 返回字典类型的UDF定义

# coding:utf8
import string

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType, ArrayType
from pyspark.sql import functions as F
if __name__ == '__main__':
    # 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.appName("test").master("local[*]").getOrCreate()
    sc = spark.sparkContext

    # 假设 有三个数字 1,2，3 我们传入数字， 返回数字所在序号对应的 字母 然后和数字结合形成dict返回
    # 比如传入 1 我们返回 {“num" : 1, "letters": "a"}

    rdd = sc.parallelize([[1], [2], [3]])
    df = rdd.toDF(["num"])

    def process(data):
        return {"num":data, "letters": string.ascii_letters[data]}

    # UDF的返回值是字典的话，需要用 StructType来接受

    udf1 = spark.udf.register("udf1", process, StructType().add("num", IntegerType(), nullable=True).\
                             add("letters", StringType(), nullable=True))
    df.selectExpr("udf1(num)").show(truncate=False)
    df.select(udf1(df['num'])).show(truncate=False)

# 输出结果
+---------+
|udf1(num)|
+---------+
|{1, b}   |
|{2, c}   |
|{3, d}   |
+---------+

+---------+
|udf1(num)|
+---------+
|{1, b}   |
|{2, c}   |
|{3, d}   |
+---------+

2 窗口函数

    开窗函数的引入是为了既显示聚集前的数据，又显示聚集后的数据。即在每一行的最后一列添加聚合函数的结果。开窗用于为定义一个窗口（这里的窗口是指运算将要操作的行的集合），它对一组值进行操作，不需要使用GROUP BY 子句 对数据进行分组，能够在同一行中同时返回基础行的列和聚合列

    聚合函数 和 开窗函数

聚合函数是将多行变成一行，count,acg....

开窗函数是将一行变成多行

聚合函数如果要显示其他的列必须将列加入到 group by中

开窗函数可以不使用group by, 直接将所有信息显示出来

    开窗函数分类

1.聚合开窗函数

聚合函数（列）OVER(选项），这里的选项可以是PARTITION BY 子句，但不可以是 ORDER BY 句子

2.排序开窗函数

排序函数（列）OVER(选项）,这里的选项可以是ORDER BY 子句，也可以是OVER（ PARTITION BY子句 ORDER BY 子句），但不可以是PARTITION BY 子句

3.分区类型NTIL的窗口函数

# coding:utf8

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType, ArrayType
from pyspark.sql import functions as F
if __name__ == '__main__':
    # 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.appName("test").master("local[*]").getOrCreate()
    sc = spark.sparkContext

    rdd = sc.parallelize([
        ("张三", "class_1", 99),
        ("张1", "class_2", 19),
        ("张2", "class_2", 19),
        ("张3", "class_2", 29),
        ("张4", "class_3", 49),
        ("张5", "class_3", 12),
        ("张6", "class_3", 14),
        ("张7", "class_3", 32),
        ("张8", "class_3", 22),
        ("张22", "class_4", 24),
        ("张11", "class_4", 49)
    ])
    schema = StructType().add("name", StringType()).\
        add("class", StringType()).\
        add("score", IntegerType())

    df = rdd.toDF(schema=schema)

    df.createTempView("stu")

    # 聚合窗口函数 的展示
    spark.sql(
        """
            SELECT *, avg(score) OVER() AS avg_score FROM stu
        """
    ).show()

    # 排序相关的 窗口函数计算
    # RANK over ,DENSE_RANK over  ROW_NUMBER over
    spark.sql("""
        SELECT *, ROW_NUMBER() OVER(ORDER BY score DESC) AS row_number_rank,
        DENSE_RANK() OVER(PARTITION BY class ORDER BY score DESC) AS dense_rank,
        RANK() OVER(order by score) AS rank
        FROM stu
    """).show()

    # NTILE
    spark.sql("""
        SELECT *, NTILE(6) OVER(ORDER BY score DESC) from stu
    """).show()

# 输出

+----+-------+-----+-----------------+
|name|  class|score|        avg_score|
+----+-------+-----+-----------------+
|张三|class_1|   99|33.45454545454545|
| 张1|class_2|   19|33.45454545454545|
| 张2|class_2|   19|33.45454545454545|
| 张3|class_2|   29|33.45454545454545|
| 张4|class_3|   49|33.45454545454545|
| 张5|class_3|   12|33.45454545454545|
| 张6|class_3|   14|33.45454545454545|
| 张7|class_3|   32|33.45454545454545|
| 张8|class_3|   22|33.45454545454545|
|张22|class_4|   24|33.45454545454545|
|张11|class_4|   49|33.45454545454545|
+----+-------+-----+-----------------+

+----+-------+-----+---------------+----------+----+
|name|  class|score|row_number_rank|dense_rank|rank|
+----+-------+-----+---------------+----------+----+
|张三|class_1|   99|              1|         1|  11|
| 张3|class_2|   29|              5|         1|   7|
| 张1|class_2|   19|              8|         2|   3|
| 张2|class_2|   19|              9|         2|   3|
| 张4|class_3|   49|              2|         1|   9|
| 张7|class_3|   32|              4|         2|   8|
| 张8|class_3|   22|              7|         3|   5|
| 张6|class_3|   14|             10|         4|   2|
| 张5|class_3|   12|             11|         5|   1|
|张11|class_4|   49|              3|         1|   9|
|张22|class_4|   24|              6|         2|   6|
+----+-------+-----+---------------+----------+----+

+----+-------+-----+-----------------------------------------------------------------------------------------------+
|name|  class|score|ntile(6) OVER (ORDER BY score DESC NULLS LAST ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)|
+----+-------+-----+-----------------------------------------------------------------------------------------------+
|张三|class_1|   99|                                                                                              1|
| 张4|class_3|   49|                                                                                              1|
|张11|class_4|   49|                                                                                              2|
| 张7|class_3|   32|                                                                                              2|
| 张3|class_2|   29|                                                                                              3|
|张22|class_4|   24|                                                                                              3|
| 张8|class_3|   22|                                                                                              4|
| 张1|class_2|   19|                                                                                              4|
| 张2|class_2|   19|                                                                                              5|
| 张6|class_3|   14|                                                                                              5|
| 张5|class_3|   12|                                                                                              6|
+----+-------+-----+-----------------------------------------------------------------------------------------------+

标签：数据库大数据 spark

本文转载自: https://blog.csdn.net/m0_55674689/article/details/128428982
版权归原作者 Darcy cui 所有，如有侵权，请联系我们删除。

SparkSQL函数定义——UDF函数，窗口函数

1 定义UDF函数

1.1 返回值是数组类型的UDF定义

1.2 返回字典类型的UDF定义

2 窗口函数

发表评论

“SparkSQL函数定义——UDF函数，窗口函数”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航