Lakeflow Designer 中的使用者自訂運算元

Lakeflow Designer 讓你能建立使用者 自訂的運算子 ，這些運算子直接出現在畫布中，與內建運算子並列。利用它們來擴充 Lakeflow Designer，加入你自己的商業邏輯、計算或整合。

使用者定義運算子有三種類型：

python-run-function：儲存在工作區中的獨立 YAML 檔案，內含內嵌 Python 程式碼。最適合用於 DataFrame 層級轉換與外部整合。權限是在工作區檔案層級管理的。
uc-udf：封裝 Unity Catalog 的純量函式。最適合用於欄位層級轉換。存取權限受 Unity 目錄權限所規範。
uc-udtf：封裝了 Unity Catalog 的表值函式。最適合用於資料表層級的轉換，例如機器學習的聚類和聚合。存取權限受 Unity 目錄權限所規範。

Feature	`python-run-function`	`uc-udf`	`uc-udtf`
使用案例範例	DataFrame 轉換、API 整合、電子郵件通知	欄位層級計算（BMI、利率）	機器學習叢集、跨列聚合
輸入	資料框架	單一價值	整個表格，逐列顯示
Output	資料框架	單一值	表格（多列）
需要 Unity 目錄功能	No	Yes	Yes
存取控管	Workspace 檔案權限	Unity Catalog 權限 (`EXECUTE`, `USE SCHEMA`)	Unity Catalog 權限 (`EXECUTE`, `USE SCHEMA`)
支援的語言	只使用 Python	SQL 或 Python 在 SQL 包裝器中	SQL 或 Python 在 SQL 包裝器中

使用者定義運算子的運作方式

使用者定義的運算子包含：

運算子邏輯：當運算子執行時執行的程式碼。這可以是內聯的 Python run() 函式（用於 python-run-function），或是 Unity 目錄函式（用於 uc-udf 和 uc-udtf）。
YAML 設定：告訴 Lakeflow Designer 如何在 UI 中呈現運算元，包括運算元名稱、描述、輸入參數、UI 元件及埠口。所有操作員類型都使用該 user-defined-operator-v0.1.0 架構。
註冊檔案：.user_defined_operators.yaml 中的一個條目，可讓 Lakeflow Designer 識別該運算子。

運算子邏輯

Python 執行函式使用者定義運算子邏輯

每個 python-run-function 算符必須定義一個 run() 函數：

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

config：使用者在 UI 中設定的值，以屬性名稱為鍵。
inputs：輸入資料框，以輸入埠 name為鍵。
spark：作用中的 SparkSession。
返回：一個將輸出埠 name 值映射到 DataFrame 的字典。

以下範例是從輸入資料框架中篩選資料列：

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

如果你的運算子需要外部 PIP 套件，請將欄位 environment 加入 YAML中：

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

UDF 與 UDTF 運算子邏輯

UC 函式可以用 SQL 或 Python 撰寫。 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 要求你定義三個關鍵方法： __init__()、、 eval()和 terminate()：

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

    def eval(self, row, id_column, columns, k):
        # Called one time per input row - accumulate data here.

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

Note

UDTF 回傳表必須有固定且明確的類型。你無法在返回設定中參考輸入欄位類型。

YAML 組態

YAML 設定告訴 Lakeflow Designer 如何在使用者介面中呈現運算子。它定義了操作員的名稱、描述、輸入參數、UI 小工具和埠口。每個設定欄位都是一個屬性，包含類型、標題及可選 x-ui 的元件提示：

config:
  type: object
  properties:
    my_param:
      type: string
      title: My Parameter
      x-ui:
        widget: input
    my_expression:
      type: string
      title: Column
      format: expression
      x-ui:
        widget: expression
        port: in
    my_number:
      type: number
      title: Count
      default: 10
      minimum: 0
      maximum: 100
  required:
    - my_param
    - my_expression

關於 YAML 架構的完整細節，包括所有元件類型與設定選項，請參閱使用者定義運算元 YAML 參考資料。

連接埠

埠口定義了你的運算元的輸入與輸出：

ports:
  input:
    - name: in
      title: Input Data
      mime: application/vnd.databricks.dataframe
      required: true
      allowMultiple: false
  output:
    - name: out
      title: Output Data

Python 的 YAML 執行函式運算子

對於 python-run-function 運算子，YAML 檔案為獨立檔案，並包含一個帶有內嵌 Python 碼的 run_function 欄位：

schema: user-defined-operator-v0.1.0
type: python-run-function
name: Filter Rows
id: filter_rows
version: '1.0.0'
description: Filters rows based on a SQL expression.
config:
  type: object
  properties:
    filter_expression:
      type: string
      title: Filter Expression
      x-ui:
        widget: input
  required:
    - filter_expression
ports:
  input:
    - name: in
      title: Input
  output:
    - name: out
      title: Output
run_function:
  type: inline
  code: |
    def run(config, inputs, spark):
        df = inputs["in"]
        filtered = df.filter(config["filter_expression"])
        return {"out": filtered}

Unity 目錄函式的 YAML

對於以 UC 為基礎的運算子，請將 YAML 設定嵌入至函式中的註解或 docstring 內。

在 SQL 中（使用 /* ... */ 註解）：

RETURN(/*
  schema: user-defined-operator-v0.1.0
  type: uc-udf
  name: Calculate BMI
  id: calculate_bmi
  version: "1.0.0"
  description: Calculates BMI from weight and height.
  config:
    type: object
    properties:
      weight_kg:
        type: string
        title: Weight (in kg)
        format: expression
        x-ui:
          widget: expression
          port: in
      height_m:
        type: string
        title: Height (in meters)
        format: expression
        x-ui:
          widget: expression
          port: in
    required:
      - weight_kg
      - height_m
  ports:
    input:
      - name: in
        title: Input Data
    output:
      - name: out
        title: Output
    */
  SELECT weight_kg / (height_m * height_m)
);

In Python（使用 """ ... """ docstring）：

AS $$
  """
  schema: user-defined-operator-v0.1.0
  type: uc-udf
  name: Calculate BMI
  id: calculate_bmi
  version: "1.0.0"
  description: Calculates BMI from weight and height.
  config:
    type: object
    properties:
      weight_kg:
        type: string
        title: Weight (in kg)
        format: expression
        x-ui:
          widget: expression
          port: in
      height_m:
        type: string
        title: Height (in meters)
        format: expression
        x-ui:
          widget: expression
          port: in
    required:
      - weight_kg
      - height_m
  ports:
    input:
      - name: in
        title: Input Data
    output:
      - name: out
        title: Output
  """

  return weight_kg / (height_m ** 2)
$$;

註冊並部署您的操作員至 Lakeflow Designer

為了讓你的操作員在 Lakeflow Designer 中出現，請將其註冊在 .user_defined_operators.yaml 一個檔案中：

工作空間層級： 將檔案放在工作區根目錄，讓操作員對所有使用者都可見。
使用者層級： 將檔案放入使用者主資料夾（/Workspace/Users/<user-name>/.user_defined_operators.yaml），讓操作員只對你可見。

此 operators: 區塊支援檔案路徑、Unity 目錄函式參考及 glob 圖案。你可以混合使用不同的入門類型：

operators:
  # File path (python-run-function operators)
  - /Workspace/Users/me/udos/my_operator.yaml
  # Glob pattern (registers all matching files)
  - /Workspace/Users/me/udos/transforms/*.yaml
  # UC function reference (uc-udf and uc-udtf operators)
  - catalog: my_catalog
    schema: my_schema
    functionName: my_function

更新或移除操作員

當你更改運算子的程式碼時，請刷新你自訂的運算子以載入變更。在選單的 「操作員」 標籤中，點擊「重新整理」圖示. .

如果操作員保持不變 version，刷新會載入更新後的程式碼。
如果操作員有一個新的 version 版本，畫布上的操作員會在重新整理後提示你升級至該版本（或保留目前版本）。

若要從 Lakeflow Designer 移除運算子，請從 .user_defined_operators.yaml 刪除其項目。對於 uc-udf 和 uc-udtf 運算子，如果您不再需要，也可以使用 DROP FUNCTION 卸除底層的 Unity Catalog 函式。

進階組態

預覽模式

Lakeflow Designer 支援在設計模式下預覽。對於呼叫外部 API 或寫入外部系統的操作員，可以新增 is_preview 設定屬性，這樣預覽時可以跳過副作用。啟用預覽模式時，使用者需明確點擊執行，才能執行帶有副作用的操作符。

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 目錄連結

對於呼叫外部 API 的 UC SQL 運算子，請使用 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() 函式來連接。詳情請參見「連接至外部 HTTP 服務」。

工作空間用戶端

對於 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

建立一個完整的 Python 執行函式使用者定義運算子

以下步驟將逐步說明如何從零開始建立 python-run-function 運算子。

步驟一：定義邏輯

在筆記本上寫下你的 run() 功能：

from typing import Dict, Any

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
    from pyspark.sql import functions as F
    df = inputs["in"]
    result = df.withColumn(config["column_name"], F.current_timestamp())
    return {"out": result}

步驟 2：測試功能

以互動方式使用範例資料測試函式：

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

result = run(
    config={"column_name": "processed_at"},
    inputs={"in": test_df},
    spark=spark
)

result["out"].show()

步驟 3：建立 YAML 配置

在 YAML 檔案中定義運算元資料、設定欄位與埠口：

schema: user-defined-operator-v0.1.0
type: python-run-function
name: Add Timestamp
id: transforms.add_timestamp
version: '1.0.0'
description: Adds a timestamp column to the input DataFrame.
config:
  type: object
  properties:
    column_name:
      type: string
      title: Column Name
      default: processed_at
      x-ui:
        widget: input
  required:
    - column_name

步驟 4：結合邏輯與 YAML系統

加入 run_function 和 ports 欄位即可建立完整的 YAML 檔案。將其儲存至你的工作區，例如 /Workspace/Users/<user-name>/udos/add_timestamp.yaml：

schema: user-defined-operator-v0.1.0
type: python-run-function
name: Add Timestamp
id: transforms.add_timestamp
version: '1.0.0'
description: Adds a timestamp column to the input DataFrame.
config:
  type: object
  properties:
    column_name:
      type: string
      title: Column Name
      default: processed_at
      x-ui:
        widget: input
  required:
    - column_name
ports:
  input:
    - name: in
      title: Input
  output:
    - name: out
      title: Output
run_function:
  type: inline
  code: |
    from typing import Dict, Any

    def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
        from pyspark.sql import functions as F
        df = inputs["in"]
        result = df.withColumn(config["column_name"], F.current_timestamp())
        return {"out": result}

步驟五：註冊營運商

將檔案路徑加入你的 .user_defined_operators.yaml 檔案：

operators:
  - /Workspace/Users/<user-name>/udos/add_timestamp.yaml

步驟 6：使用 Lakeflow Designer 中的運算元

打開 Lakeflow Designer，並確認操作符是否出現在操作符調色盤中。把它拖到畫布上，連接輸入，設定欄位名稱，然後執行預覽。

建立完整的 UC 使用者定義運算元

以下步驟將逐步說明如何建立基於 uc-udf UC 的運算元。

步驟一：定義邏輯

在筆記本中撰寫並測試你的函式邏輯：

def double_value(input_value: float) -> float:
    if input_value is None:
        return None
    return input_value * 2

步驟 2：建立 YAML 配置

定義運算元資料、設定欄位與埠口：

schema: user-defined-operator-v0.1.0
type: uc-udf
name: Double Value
id: math.double_value
version: '1.0.0'
description: Doubles the input value
config:
  type: object
  properties:
    input_value:
      type: string
      title: Input Value
      format: expression
      x-ui:
        widget: expression
        port: input_data
  required:
    - input_value
ports:
  input:
    - name: input_data
      title: Input
  output:
    - name: out
      title: Output

步驟 3：結合邏輯與 YAML

建立 Unity 目錄函式，將 YAML 嵌入為文件字串：

CREATE OR REPLACE FUNCTION main.my_schema.double_value(input_value DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
AS $$
  """
  schema: user-defined-operator-v0.1.0
  type: uc-udf
  name: Double Value
  id: math.double_value
  version: "1.0.0"
  description: Doubles the input value
  config:
    type: object
    properties:
      input_value:
        type: string
        title: Input Value
        format: expression
        x-ui:
          widget: expression
          port: input_data
    required:
      - input_value
  ports:
    input:
      - name: input_data
        title: Input
    output:
      - name: out
        title: Output
  """

  def double_value(input_value: float) -> float:
      if input_value is None:
          return None
      return input_value * 2

  return double_value(input_value)
$$

步驟四：測試功能

SELECT main.my_schema.double_value(5) AS result;
-- Should return: 10

步驟五：註冊營運商

將 Unity Catalog 函式參考新增至您的 .user_defined_operators.yaml 檔案：

operators:
  - catalog: main
    schema: my_schema
    functionName: double_value

步驟 6：使用 Lakeflow Designer 中的運算元

打開 Lakeflow Designer，並確認操作符是否出現在操作符調色盤中。把它拖到畫布上，連接輸入，然後執行預覽。

故障排除

Issue	解決方案
Operator 不會出現在 Lakeflow Designer 裡。	檢查 `.user_defined_operators.yaml` 是否存在，並且其中列出了你的函式或檔案路徑。操作 `python-run-function` 員請確認檔案路徑及 YAML 檔案是否可存取。
架構驗證失敗。	請在 `https://your-workspace.cloud.databricks.com/static/schemas/user-defined-operator-v0.1.0.json` 根據官方結構描述驗證您的 YAML。
權限遭拒。	對於以 UC 為基礎的運算子，請確認使用者對函式具有 `EXECUTE`，且對結構描述具有 `USE SCHEMA`。對於 `python-run-function` 操作員，請確認使用者具有 YAML 檔案的讀取權限。
`python-run-function` 運算子在執行時失敗。	檢查函數簽名是否 `run()` 符合 `def run(config, inputs, spark)`。確認程式碼中的埠名是否與 YAML 相符，且回傳字典鍵是否符合輸出埠 `name` 值。
UDTF 回傳的型別不正確。	UDTF 回傳類型必須明確;你無法參考輸入欄位類型。

權限

許可	Purpose
對的`.user_defined_operators.yaml`。	找出操作者。
對 YAML 檔案的讀取權限（`python-run-function`僅限此權限）。	載入運算元定義。
在 Unity Catalog 函式（僅適用於以 UC 為基礎的運算子）上具有 EXECUTE 權限。	執行運算子。
USE SCHEMA 在結構描述上（僅限以 UC 為基礎的運算子）。	存取建立函式的結構模式。
其他權限	根據您的營運商，使用者可能需要其他權限。例如， `USE CONNECTION` 在 Unity 目錄連線中用於 HTTP API 呼叫。

其他資源

請探索以下教學：

Example	類型	Description
Gmail 電子郵件寄件人	`python-run-function`	透過 Gmail 以 CSV 電子郵件附件形式傳送 DataFrame 資料。
複利計算器	`uc-udf`	請使用複利公式計算未來投資價值。
K-Means 叢集	`uc-udtf`	使用 scikit-learn 將資料分割成叢集。
發送 Slack 訊息	`uc-udf`	透過 API 向 Slack 頻道發送通知。
所有 UI 小工具	`uc-udf`	參考運算子展示所有可用的 UI 元件。

欲完整參考 YAML 架構，請參見使用者定義運算元 YAML 參考。

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-06-25