为Lakeflow Designer自定义操作符编写逻辑

算子逻辑是自定义算子运行的代码。对于python-run-function算符来说，它是一个内联的 Python run() 函数。对于 uc-udf 或 uc-udtf 操作符，它是用 SQL 或 Python 编写的 Unity Catalog 函数。

编写逻辑是构建共享自定义操作符的一个步骤。写好并测试逻辑后，用 YAML 描述操作符，让 Lakeflow Designer 知道如何在界面中显示，然后注册它，使其出现在 操作符 面板中。要查看同一个运算符的所有步骤，请参阅教程，例如教程：复利计算器运算符。

编写 python-run-function 运算符

每个 python-run-function 运算符都必须定义一个 run() 函数：

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:

config：UI 中用户配置的值，按属性名称键键。
inputs：输入 DataFrame，以输入端口 name为键。
spark：当前活动的 SparkSession。
返回：字典将输出端口 name 值映射到 DataFrames。

以下示例筛选输入数据帧中的行：

def run(config, inputs, spark):
    df = inputs["in"]
    filtered = df.filter(config["filter_expression"])
    return {"out": filtered}

在注册前测试该函数，请在笔记本中调用它并附有示例数据：

test_df = spark.createDataFrame(
    [("Alice", 100), ("Bob", 200)],
    ["name", "amount"]
)

result = run(
    config={"filter_expression": "amount > 100"},
    inputs={"in": test_df},
    spark=spark
)

result["out"].show()

如果操作员需要外部 pip 包，请将 environment 字段添加到 YAML：

environment:
  environment_version: '4'
  dependencies:
    - requests==2.31.0
    - beautifulsoup4==4.12.0

编写 UDF 或 UDTF 运算符

你可以用SQL或Python编写Unity Catalog函数。 Python函数包装在 SQL CREATE FUNCTION 语句中：

SQL 函数：

CREATE OR REPLACE FUNCTION my_catalog.my_schema.calculate_bmi(weight_kg DOUBLE, height_m DOUBLE)
RETURNS DOUBLE
LANGUAGE SQL
RETURN
  SELECT weight_kg / (height_m * height_m);

Python 函数（包装在 SQL 中）：

CREATE OR REPLACE FUNCTION my_catalog.my_schema.calculate_bmi(weight_kg DOUBLE, height_m DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
AS $$
  return weight_kg / (height_m ** 2)
$$;

UDF 一次处理单个值并返回计算值。 UDTF 逐行处理表中的数据，并可在所有行之间保持状态。使用 uc-udf 进行列级转换，使用 uc-udtf 进行机器学习聚类或聚合等操作。

UDTF 处理类定义了 eval()，它会针对每个输入行运行。可选地，它还定义 __init__() 用于设置状态，以及 terminate() 用于在所有行处理完后生成最终结果。请参阅 Unity 目录中的 Python 用户定义的表函数（UDF）。

class MyOperator:
    def __init__(self):
        # Optional. Called before processing - initialize any values needed.
        ...

    def eval(self, row):
        # Required. Called one time per input row - process or accumulate data here.
        ...

    def terminate(self):
        # Optional. Called after all rows - perform final calculations and yield results.
        ...

Note

UDTF 返回表必须具有固定的显式类型。不能在返回配置中引用输入列类型。

在预览期间跳过副效应

在设计模式下，Lakeflow Designer 支持预览。对于调用外部 API 或写入外部系统的运算符，请添加配置 is_preview 属性，以便在预览期间跳过副作用。用户点击“运行”即可运行带有副作用的操作符。内置的 Python 操作符不需要这个属性，因为它会自动读取config["is_preview"]。请参见 Python 预览模式。

config:
  type: object
  properties:
    is_preview:
      type: boolean
      format: is_preview
      default: false

Lakeflow Designer 会在预览期间自动将此值设置为 true 。在逻辑中进行判断以跳过副作用：

# In a python-run-function
if config.get("is_preview"):
    return {"out": inputs["in"]}

-- In a UC function (SQL)
CASE WHEN is_preview THEN 'preview' ELSE /* actual work */ END

使用 Unity Catalog HTTP 连接调用外部 API

对于基于 SQL Unity Catalog 函数调用外部 API 的自定义操作符，使用 Unity Catalog 的 HTTP 连接安全地存储凭证：

CREATE CONNECTION my_api_connection TYPE HTTP OPTIONS (
  host 'https://api.example.com',
  port '443',
  base_path '/v1/',
  bearer_token 'your-token-here'
);

然后在 SQL UDF 中使用该连接和 http_request() 函数。

使用 WorkspaceClient 访问工作区资源

对于 python-run-function 运算符，可以使用 Azure Databricks WorkspaceClient 访问工作区资源和外部 API：

def run(config, inputs, spark):
    from databricks.sdk import WorkspaceClient
    w = WorkspaceClient()
    # Use w to access workspace resources

其他资源

上次更新时间为 2026-11-02