Apache Doris Python UDF：让 SQL 直接调用 Python 生态，支撑 Agent 时代复杂业务逻辑

SelectDB技术团队

124人浏览 · 2026-07-01 17:38:47

SelectDB技术团队 · 2026-07-01 17:38:47 发布

随着 AI 应用和实时分析场景深入，进入数据平台的不再只是结构化业务表。日志、JSON、文本内容、行为事件、模型推理结果等半结构化和非结构化数据，正在成为实时分析的常见对象。

与此同时，分析链路中要完成的工作也在变化：它不再只是 COUNT、SUM、GROUP BY，还包括规则判断、字段解析、特征加工、标签抽取、模型打分等更复杂的业务逻辑。

这些逻辑往往更适合用 Python 实现。但如果把数据导出到外部脚本或服务中处理，就会带来链路拉长、时效下降、排查困难和治理复杂等问题。

Doris Python UDF 要解决的正是这个割裂：让开发者可以在 SQL 中创建并调用 Python 函数，将 Pandas、PyArrow 等 Python 生态能力引入 Doris 查询链路，在数据不离开分析链路的前提下完成更复杂的数据处理和业务计算。

1-传统 vs udf.png

SelectDB 也已将这一能力纳入商业化产品体系，面向企业生产环境提供更完整的运维、稳定性、安全合规和技术支持能力。

像调用 SQL 函数一样调用 Python UDF

使用 Doris Python UDF 的心智模型非常简单：准备 Python 环境，声明函数，然后在 SQL 中直接调用。

2-Python UDF 使用总览.png

前置条件：在创建 Python UDF/UDAF/UDTF 前，需要在所有 BE 节点开启 Python UDF 相关配置，并在目标 Python 环境中安装 pandas 与 pyarrow。Python UDF Server 日志可在 output/be/log/python_udf_output.log 中查看。

下面以支付金额风险等级评估为例，创建一个 Python UDF：

DROP FUNCTION IF EXISTS py_risk_level(DOUBLE);

CREATE FUNCTION py_risk_level(DOUBLE)
RETURNS STRING
PROPERTIES (
    "type" = "PYTHON_UDF",
    "symbol" = "evaluate",
    "runtime_version" = "3.12.11",
    "always_nullable" = "true",
    "volatility" = "immutable"
)
AS $$
def evaluate(amount):
    if amount is None:
        return None
    if amount >= 10000:
        return "high"
    if amount >= 1000:
        return "medium"
    return "low"
$$;

创建完成后，该函数即可像 Doris 内置函数一样使用：

SELECT
    user_id,
    amount,
    py_risk_level(amount) AS risk_level
FROM payment_events
WHERE dt = '2026-06-17'
ORDER BY user_id;

开发者可以把已有 Python 逻辑直接嵌入 SQL 查询链路，在 Doris 内完成数据处理、规则判断和特征加工。

Python UDF 核心特点

3-Python UDF 核心特点.png

基于 Arrow RecordBatch 的批量执行

Python UDF 既要具备 Python 的灵活性，也需要尽可能降低跨语言、跨进程调用带来的额外开销。

Doris Python UDF 采用列式批量处理机制。执行过程中，Doris BE 会将输入数据组织为 Arrow RecordBatch，并通过 Arrow Flight 高效传输至独立的 Python Server。Python 函数完成批量计算后，结果再以列式数据形式返回 Doris 查询链路。

这种方式避免了传统逐行调用造成的频繁进程切换和序列化开销，使 Python 扩展能力能够与 Doris 的列式执行框架保持一致，在支持复杂业务逻辑的同时，尽可能保持查询执行效率。

4-arrow 批量执行.png

支持 Pandas Series 向量化计算

对于字符串处理、特征计算、字段转换、分桶映射等列式处理场景，Python UDF 支持基于 Pandas Series 的向量化实现。

例如，可以使用 Pandas 对金额进行分桶：

CREATE FUNCTION py_amount_bucket(DOUBLE)
RETURNS INT
PROPERTIES (
    "type" = "PYTHON_UDF",
    "symbol" = "evaluate",
    "runtime_version" = "3.10.12",
    "always_nullable" = "true",
    "volatility" = "immutable"
)
AS $$
import pandas as pd

# 显式声明pd.Series类型，使用向量化实现
def evaluate(amount: pd.Series) -> pd.Series:
    return pd.cut(
        amount,
        bins=[-float("inf"), 100, 1000, 10000, float("inf")],
        labels=[0, 1, 2, 3]
    ).astype("Int64")
$$;

相比在 Python 中逐行循环处理，向量化计算可以更好利用 Pandas 底层能力，减少解释器循环开销，适合大批量数据转换和特征加工场景。

完整支持 UDF、UDAF、UDTF

Doris Python 扩展能力覆盖三类函数形态：

5-完整支持生态.png

同一套 Python 扩展框架能够覆盖标量计算、聚合计算和展开型处理，降低不同业务逻辑接入 Doris 的复杂度。

支持内联与模块化加载

Doris Python UDF 支持灵活的代码组织方式。

方式一：内联方式。对于简单函数，可以直接将 Python 代码写在 CREATE FUNCTION 语句中，适合快速验证和小规模试验：

CREATE FUNCTION py_add_one(INT)
RETURNS INT
PROPERTIES (
    "type" = "PYTHON_UDF",
    "symbol" = "evaluate",
    "runtime_version" = "3.10.12",
    "volatility" = "immutable"
)
AS $$
def evaluate(x):
    return None if x is None else x + 1
$$;

方式二：模块方式。对于复杂函数，可以将 Python 代码打成 ZIP 包，并通过 file 与 symbol 指定模块入口：

CREATE FUNCTION py_add_one(INT)
RETURNS INT
PROPERTIES (
    "type" = "PYTHON_UDF",
    "file" = "file:///opt/doris/udf/math_ops.zip",
    "symbol" = "math_ops.add_one",
    "runtime_version" = "3.10.12",
    "volatility" = "immutable"
);

这种方式更适合团队协作、代码评审、依赖管理和版本发布，也便于将生产级 Python 逻辑稳定接入 Doris。