Python UDAF

Python UDAF（User Defined Aggregation Function）是一种用户自定义的功能，它允许你在Apache Spark中编写复杂的聚合操作，而不仅仅限于简单的求和、平均等基础操作。UDAFs提供了一种处理分布式数据集的强大工具，它们通常在每个分区上应用函数，然后将结果合并到最终汇总值。

UDAF的特点包括：
1. 分区操作：在Spark集群上并行执行，对每个分区的数据独立计算，提高了性能。
2. 非确定性和状态管理：支持非确定性的输入，例如集合操作，并维护中间状态。
3. 结果处理：最后会将各个分区的结果归约并返回单一汇总值。

使用Python UDAF的例子中，你可能会创建一个`CountDistinct`函数，用于计算每一组键的唯一值数量，或者一个`WindowFunction`来处理滑动窗口内的数据统计。

相关问题

阿里云 python 构建UDAF，传入一个列名col和percentile，实现计算某列的分位值

在阿里云MaxCompute中，你可以使用Python构建一个用户定义聚合函数(UDAF)来计算指定列的百分位数。例如，假设你想计算名为`col`的一列的第`percentile`百分位值，可以按照以下步骤操作：

首先，你需要定义一个类，继承自`pyodps.udf.UDAF`类：

from pyodps.udf import annotate
from collections import defaultdict

@annotate('quantile', 'double')
class PercentileUDAF(object):
    def __init__(self, col_name, percentile):
        self.col_name = col_name
        self.percentile = percentile
        self.values = defaultdict(list)
        self.rank_sum = 0

    # 初始化阶段，存储所有输入值及其索引
    def initialize(self, ctx):
        for i, row in enumerate(ctx.get_iterator()):
            value = row[self.col_name]
            self.values[value].append(i)

    # 迭代阶段，将新值和索引入队列
    def iterate(self, ctx, value):
        if value is not None:
            rank = len(self.values[value])
            self.values[value].append(rank)
            self.rank_sum += rank

    # 结束阶段，计算并返回百分位值
    def terminate(self, ctx):
        total_count = ctx.get_total_size()
        values = list(self.values.values())
        for val_list in values:
            sorted_val_list = sorted(val_list)
            idx = int((self.rank_sum / total_count) * len(sorted_val_list))
            if idx < len(sorted_val_list):
                return sorted_val_list[idx]

# 使用示例
percentile_udaf = PercentileUDAF('col', 0.5)  # 计算50%分位数（中位数）

然后，在SQL查询中调用这个UDAF：

SELECT quantile(col, @percentile) FROM your_table;

这里`@percentile`是一个系统变量，你需要预先设置它的值为你想要计算的百分比。

用Python分别写一个udf、udaf、udtf函数

1. UDF（用户自定义函数）示例：

from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType

def square(x):
    return x * x

square_udf = udf(square, IntegerType())

df = spark.createDataFrame([(1,), (2,), (3,), (4,), (5,)], ["num"])
df.withColumn("square", square_udf("num")).show()

输出结果：

+---+------+
|num|square|
+---+------+
|  1|     1|
|  2|     4|
|  3|     9|
|  4|    16|
|  5|    25|
+---+------+

2. UDAF（用户自定义聚合函数）示例：

from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType, StructType, StructField

def my_average(values):
    return sum(values) / len(values)

schema = StructType([StructField("value", IntegerType())])
my_average_udaf = udf(my_average, IntegerType())

df = spark.createDataFrame([(1,), (2,), (3,), (4,), (5,)], ["num"])
df.agg(my_average_udaf(df.num)).show()

输出结果：

+-----------+
|my_average(num)|
+-----------+
|          3|
+-----------+

3. UDTF（用户自定义表生成函数）示例：

from pyspark.sql.functions import explode
from pyspark.sql.types import ArrayType, StringType

def split_words(text):
    return text.split()

split_words_udf = udf(split_words, ArrayType(StringType()))

df = spark.createDataFrame([("hello world",), ("spark is awesome",)], ["text"])
df.select(explode(split_words_udf(df.text))).show()

输出结果：

+------+
|   col|
+------+
| hello|
| world|
| spark|
|    is|
|awesome|
+------+