创建UDAF
将一个聚合函数（UDAF）注册到LakeFormation。聚合函数对多行数据进行聚合计算。除udf_type固定为udaf外，参数与create_udf完全一致。
前提条件
- 已完成SDK安装。具体操作，请参见Catalog SDK获取与安装。
- 已准备好UDF可调用对象（函数、函数类或类实例），并标注了类型注解。
函数签名
client.create_udaf(
udaf, /, *,
database=None,
name,
signature=None,
imports=None,
runtime_env=None,
description=None,
pretty_name=None,
is_strict=True,
volatility="VOLATILE",
is_batch=False,
batch_size=None,
cpu_arch=None,
cpus=None,
memory=None,
device_type=None,
devices=None,
) -> Function 参数说明
| 参数 | 是否必选 | 类型 | 默认值 | 说明 |
|---|---|---|---|---|
| udaf | 是 | Any（位置参数） | - | udaf可调用对象（函数或可调用类）。 |
| database | 否 | str | None | default_database | 目标数据库名；为None时使用客户端默认数据库。 |
| name | 是 | str | - | 函数名，须以字母开头，仅含字母/数字/下划线，长度 3–128。 |
| signature | 否 | Signature | None | None | 显式函数签名；为None时由UDF注解自动推断。 关于signature的更多信息，请参见signature说明。 |
| imports | 否 | str | Iterable[str] | None | None | 需要打入UDF包的文件/目录路径或URL列表。 |
| runtime_env | 否 | str | Mapping | None | None | 运行时环境配置（python_version、pypi 依赖等），可传 JSON 字符串或字典。 关于runtime_env的更多信息，请参见runtime_env说明。 |
| description | 否 | str | None | None | 函数描述。 |
| pretty_name | 否 | str | None | None | 展示名；为None时使用name。 |
| is_strict | 否 | bool | True | 函数是否严格（输入为NULL时是否直接返回NULL）。 |
| volatility | 否 | str | “VOLATILE” | 函数易变性：IMMUTABLE/STABLE/VOLATILE。 |
| is_batch | 否 | bool | False | 是否启用批量处理模式。 |
| batch_size | 否 | int | None | None | 批量处理每批大小；is_batch=True 时建议提供，须为正整数。 |
| cpu_arch | 否 | str | None | None | CPU架构，取值arm或x86。 |
| cpus | 否 | float | None | None | CPU核数，最小0.33。 |
| memory | 否 | int | None | None | 内存大小，单位MB，最小1。 |
| device_type | 否 | str | None | None | 设备类型，例如npu、gpu。 |
| devices | 否 | float | None | None | 设备数量，需为正数。 |
返回值
返回Function对象。
示例代码
class WeightedSum:
"""UDAF：加权求和 sum(value * weight)。
accumulate 逐行累积中间态（sum_vw），merge 合并分区，
finish 返回最终聚合结果。
"""
def __init__(self):
self._agg_state = 0.0
@property
def aggregate_state(self) -> float:
return self._agg_state
def accumulate(self, value: int, weight: int) -> None:
"""逐行累积：value * weight 求和"""
if value is not None and weight is not None:
self._agg_state += value * weight
def merge(self, other_state: float) -> None:
"""合并其他分区的中间态"""
self._agg_state += other_state
def finish(self) -> float:
"""返回最终聚合结果"""
return self._agg_state
func = client.create_agg_function(
WeightedSum,
database=database,
name="weighted_sum_udaf",
signature=sig,
comment="UDAF: 加权求和 sum(value * weight)",
) 
