SparkSQL 定義UDF函數(shù)

方式1語法：
udf對象 = sparksession.udf.register(參數(shù)1，參數(shù)2，參數(shù)3）
參數(shù)1：UDF名稱，可用于SQL風(fēng)格
參數(shù)2：被注冊成UDF的方法名
參數(shù)3：聲明UDF的返回值類型
udf對象： 返回值對象，是一個UDF對象，可用于DSL風(fēng)格
方式2語法：
udf對象 = F.udf(參數(shù)1， 參數(shù)2)
參數(shù)1：被注冊成UDF的方法名
參數(shù)2：聲明UDF的返回值類型
udf對象： 返回值對象，是一個UDF對象，可用于DSL風(fēng)格
其中F是：
from pyspark.sql import functions as F
其中，被注冊成UDF的方法名是指具體的計算方法，如：
def add(x, y): x + y
add就是將要被注冊成UDF的方法名

??

# coding:utf8
import timefrom pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType
import pandas as pd
from pyspark.sql import functions as Fif __name__ == '__main__':# 0. 構(gòu)建執(zhí)行環(huán)境入口對象SparkSessionspark = SparkSession.builder.\appName("test").\master("local[*]").\config("spark.sql.shuffle.partitions", 2).\getOrCreate()sc = spark.sparkContext# 構(gòu)建一個RDDrdd = sc.parallelize([1, 2, 3, 4, 5, 6, 7]).map(lambda x:[x])df = rdd.toDF(["num"])# TODO 1: 方式1 sparksession.udf.register(), DSL和SQL風(fēng)格均可以使用# UDF的處理函數(shù)def num_ride_10(num):return num * 10# 參數(shù)1: 注冊的UDF的名稱, 這個udf名稱, 僅可以用于 SQL風(fēng)格# 參數(shù)2: UDF的處理邏輯, 是一個單獨的方法# 參數(shù)3: 聲明UDF的返回值類型, 注意: UDF注冊時候, 必須聲明返回值類型, 并且UDF的真實返回值一定要和聲明的返回值一致# 返回值對象: 這是一個UDF對象, 僅可以用于 DSL 語法# 當(dāng)前這種方式定義的UDF, 可以通過參數(shù)1的名稱用于SQL風(fēng)格, 通過返回值對象用戶DSL風(fēng)格udf2 = spark.udf.register("udf1", num_ride_10, IntegerType())# SQL風(fēng)格中使用# selectExpr 以SELECT的表達式執(zhí)行, 表達式 SQL風(fēng)格的表達式(字符串)# select方法, 接受普通的字符串字段名, 或者返回值是Column對象的計算df.selectExpr("udf1(num)").show()# DSL 風(fēng)格中使用# 返回值UDF對象 如果作為方法使用, 傳入的參數(shù) 一定是Column對象df.select(udf2(df['num'])).show()# TODO 2: 方式2注冊, 僅能用于DSL風(fēng)格udf3 = F.udf(num_ride_10, IntegerType())df.select(udf3(df['num'])).show()df.selectExpr("udf3(num)").show()

?

# coding:utf8
import timefrom pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType, ArrayType
import pandas as pd
from pyspark.sql import functions as Fif __name__ == '__main__':# 0. 構(gòu)建執(zhí)行環(huán)境入口對象SparkSessionspark = SparkSession.builder.\appName("test").\master("local[*]").\config("spark.sql.shuffle.partitions", 2).\getOrCreate()sc = spark.sparkContext# 構(gòu)建一個RDDrdd = sc.parallelize([["hadoop spark flink"], ["hadoop flink java"]])df = rdd.toDF(["line"])# 注冊UDF, UDF的執(zhí)行函數(shù)定義def split_line(data):return data.split(" ")  # 返回值是一個Array對象# TODO1 方式1 構(gòu)建UDFudf2 = spark.udf.register("udf1", split_line, ArrayType(StringType()))# DLS風(fēng)格df.select(udf2(df['line'])).show()# SQL風(fēng)格df.createTempView("lines")spark.sql("SELECT udf1(line) FROM lines").show(truncate=False)# TODO 2 方式2的形式構(gòu)建UDFudf3 = F.udf(split_line, ArrayType(StringType()))df.select(udf3(df['line'])).show(truncate=False)

?

# coding:utf8
import string
import timefrom pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType, ArrayType
import pandas as pd
from pyspark.sql import functions as Fif __name__ == '__main__':# 0. 構(gòu)建執(zhí)行環(huán)境入口對象SparkSessionspark = SparkSession.builder.\appName("test").\master("local[*]").\config("spark.sql.shuffle.partitions", 2).\getOrCreate()sc = spark.sparkContext# 假設(shè) 有三個數(shù)字  1 2 3  我們傳入數(shù)字 ,返回數(shù)字所在序號對應(yīng)的 字母 然后和數(shù)字結(jié)合形成dict返回# 比如傳入1 我們返回 {"num":1, "letters": "a"}rdd = sc.parallelize([[1], [2], [3]])df = rdd.toDF(["num"])# 注冊UDFdef 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)

??

?SparkSQL 使用窗口函數(shù)

# coding:utf8
# 演示sparksql 窗口函數(shù)(開窗函數(shù))
import string
from pyspark.sql import SparkSession
# 導(dǎo)入StructType對象
from pyspark.sql.types import ArrayType, StringType, StructType, IntegerType
import pandas as pd
from pyspark.sql import functions as Fif __name__ == '__main__':spark = SparkSession.builder. \appName("create df"). \master("local[*]"). \config("spark.sql.shuffle.partitions", "2"). \getOrCreate()
sc = spark.sparkContext
rdd = sc.parallelize([('張三', 'class_1', 99),('王五', 'class_2', 35),('王三', 'class_3', 57),('王久', 'class_4', 12),('王麗', 'class_5', 99),('王娟', 'class_1', 90),('王軍', 'class_2', 91),('王俊', 'class_3', 33),('王君', 'class_4', 55),('王珺', 'class_5', 66),('鄭穎', 'class_1', 11),('鄭輝', 'class_2', 33),('張麗', 'class_3', 36),('張張', 'class_4', 79),('黃凱', 'class_5', 90),('黃開', 'class_1', 90),('黃愷', 'class_2', 90),('王凱', 'class_3', 11),('王凱杰', 'class_1', 11),('王開杰', 'class_2', 3),('王景亮', 'class_3', 99)
])
schema = StructType().add("name", StringType()). \add("class", StringType()). \add("score", IntegerType())
df = rdd.toDF(schema)
# 窗口函數(shù)只用于SQL風(fēng)格, 所以注冊表先
df.createTempView("stu")
# TODO 聚合窗口
spark.sql("""
SELECT *, AVG(score) OVER() AS avg_score FROM stu
""").show()
# SELECT *, AVG(score) OVER() AS avg_score FROM stu 等同于
# SELECT * FROM stu
# SELECT AVG(score) FROM stu
# 兩個SQL的結(jié)果集進行整合而來
spark.sql("""
SELECT *, AVG(score) OVER(PARTITION BY class) AS avg_score FROM stu
""").show()
# SELECT *, AVG(score) OVER(PARTITION BY class) AS avg_score FROM stu 等同于
# SELECT * FROM stu
# SELECT AVG(score) FROM stu GROUP BY class
# 兩個SQL的結(jié)果集進行整合而來
# TODO 排序窗口
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()
# TODO NTILE
spark.sql("""
SELECT *, NTILE(6) OVER(ORDER BY score DESC) FROM stu
""").show()

SparkSQL支持UDF和UDAF定義，但在Python中，暫時只能定義UDF
UDF定義支持2種方式, 1:使用SparkSession對象構(gòu)建. 2: 使用functions包中提供的UDF API構(gòu)建. 要注意, 方式1可用DSL和SQL風(fēng)格, 方式2 僅可用于DSL風(fēng)格
SparkSQL支持窗口函數(shù)使用, 常用SQL中的窗口函數(shù)均支持, 如聚合窗口\排序窗口\NTILE分組窗口等

?