Uživatelem definované operátory v Návrháři Lakeflow

Návrhář Lakeflow umožňuje vytvářet uživatelem definované operátory , které se zobrazují přímo na plátně spolu s integrovanými operátory. Použijte je k rozšíření Návrháře Lakeflow vlastní obchodní logikou, výpočty nebo integracemi.

Existují tři typy uživatelem definovaných operátorů:

python-run-function: Samostatný soubor YAML s vloženým kódem Pythonu uložený v pracovním prostoru. Nejvhodnější pro transformace na úrovni datového rámce a externí integrace. Oprávnění se spravují na úrovni souboru pracovního prostoru.
uc-udf: Zabalí skalární funkci katalogu Unity. Nejvhodnější pro transformace na úrovni sloupců. Přístup se řídí oprávněními katalogu Unity.
uc-udtf: Zabalí funkci tabulky v katalogu Unity. Nejvhodnější pro transformace na úrovni tabulky, jako je clustering ML a agregace. Přístup se řídí oprávněními katalogu Unity.

funkce	`python-run-function`	`uc-udf`	`uc-udtf`
Příklad případu použití	Transformace datového rámce, integrace rozhraní API, e-mailová oznámení	Výpočty na úrovni sloupců (BMI, úrokové sazby)	Shlukování pomocí ML, agregace přes řádky
Vstup	Datové rámce	Jednotlivé hodnoty	Celá tabulka, řádek po řádku
Output	Datové rámce	Jedna hodnota	Tabulka (více řádků)
Vyžaduje funkci Katalogu Unity.	Ne	Ano	Ano
Zásady správného řízení přístupu	Oprávnění k souborům pracovního prostoru	Oprávnění katalogu Unity (`EXECUTE`, `USE SCHEMA`)	Oprávnění katalogu Unity (`EXECUTE`, `USE SCHEMA`)
Podporované jazyky	Pouze Python	SQL nebo Python v obálkě SQL	SQL nebo Python v obálkě SQL

Jak fungují uživatelem definované operátory

Uživatelem definovaný operátor se skládá z:

Logika operátoru: Kód, který se spustí při spuštění operátoru. Může se jednat o vloženou funkci Python run() (pro python-run-function) nebo funkci katalogu Unity (pro uc-udf a uc-udtf).
Konfigurace YAML: Informuje Návrháře Lakeflow, jak prezentovat operátora v uživatelském rozhraní, včetně názvu operátora, popisu, vstupních parametrů, widgetů uživatelského rozhraní a portů. Všechny typy operátorů používají schéma user-defined-operator-v0.1.0.
Registrační soubor: Záznam v .user_defined_operators.yaml, který nástroji Lakeflow Designer umožňuje rozpoznat operátor.

Logika operátoru

Logika funkce run uživatelsky definovaného operátoru v Pythonu

Každý python-run-function operátor musí definovat run() funkci:

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

config: Uživatelem nakonfigurované hodnoty z uživatelského rozhraní s klíčem podle názvu vlastnosti.
inputs: Vstupní datové rámce, klíčované vstupním portem name.
spark: Aktivní SparkSession.
Návratová hodnota: Slovník, který mapuje hodnoty výstupních portů name na objekty DataFrame.

Následující příklad filtruje řádky ze vstupního datového rámce:

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

Pokud váš operátor vyžaduje externí balíčky pip, přidejte environment pole do YAML:

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

Logika operátoru UDF a UDTF

Funkce UC můžete psát v SQL nebo Python. Python funkce jsou zabalené v příkazu SQL CREATE FUNCTION:

Funkce 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);

funkce Python (zabalená v 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 zpracovávají vždy jednu hodnotu a vracejí vypočtenou hodnotu. Tabulkové funkce definované uživatelem zpracovávají tabulky po řádcích a mohou udržovat stav napříč všemi řádky tabulky. Používá se uc-udf pro transformace na úrovni sloupců a uc-udtf pro operace, jako je clustering ML nebo agregace.

Kromě toho UDTF vyžadují definovat tři klíčové metody: __init__(), eval() a 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

Návratové tabulky UDTF musí obsahovat pevné explicitní typy. Ve návratové konfiguraci nelze odkazovat na vstupní typy sloupců.

Konfigurace YAML

Konfigurace YAML říká Návrháři Lakeflow, jak prezentovat operátor v uživatelském rozhraní. Definuje název, popis, vstupní parametry, widgety uživatelského rozhraní a porty operátora. Každé konfigurační pole je vlastnost s typem, názvem a volitelnými x-ui nápovědami widgetu:

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

Úplné podrobnosti o schématu YAML, včetně všech typů widgetů a možností konfigurace, naleznete v referenční dokumentaci YAML uživatelem definovaného operátora.

Přístavy

Porty definují vstupy a výstupy pro operátor:

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

YAML pro operátory funkce run v Pythonu

Pro operátory python-run-function je soubor YAML samostatný a obsahuje pole run_function s vloženým kódem Python:

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}

YAML pro funkce katalogu Unity

Pro operátory založené na UC vložte konfiguraci YAML jako komentář nebo docstring do funkce.

V SQL (použijte /* ... */ komentář):

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 (použijte """ ... """ 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)
$$;

Zaregistrujte a nasaďte operátor do aplikace Lakeflow Designer

Aby se váš operátor zobrazil v Návrháři Lakeflow, zaregistrujte ho .user_defined_operators.yaml v souboru:

Úroveň pracovního prostoru: Umístěte soubor do kořenového adresáře pracovního prostoru, aby byl operátor viditelný pro všechny uživatele.
Úroveň uživatele: Umístěte soubor do domovské složky uživatele (/Workspace/Users/<user-name>/.user_defined_operators.yaml), aby se operátory zobrazovaly jenom vám.

Oddíl operators: podporuje cesty k souborům, odkazy na funkce katalogu Unity a vzory globů. Typy položek můžete kombinovat:

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

Aktualizace nebo odebrání operátoru

Když změníte kód operátora, aktualizujte uživatelem definované operátory, aby se změna načetla. Na kartě Operátory v nabídce klepněte na tlačítko Aktualizovat ikonu. .

Pokud operátor zachová stejný version, aktualizace načte aktualizovaný kód.
Pokud má operátor nový version, operátor na plátně vás vyzve k upgradu (nebo zachování aktuální verze) po aktualizaci.

Chcete-li odebrat operátor z Návrháře Lakeflow, odstraňte jeho položku z .user_defined_operators.yaml. U operátorů uc-udf a uc-udtf můžete také odstranit podkladovou funkci v Unity Catalogu pomocí DROP FUNCTION, pokud ji už nepotřebujete.

Pokročilé konfigurace

Režim náhledu

Návrhář Lakeflow podporuje náhledy v režimu návrhu. Pro operátory, které volají externí rozhraní API nebo zapisují do externích systémů, přidejte konfigurační vlastnost is_preview, abyste při náhledu mohli vynechat vedlejší účinky. Pokud je povolený režim náhledu, uživatelé musí explicitně kliknout na Spustit , aby spustili operátor s vedlejšími účinky.

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

Lakeflow Designer tuto hodnotu během náhledu automaticky nastaví na true. Zkontrolujte to ve své logice, abyste přeskočili vedlejší účinky:

# 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

Připojení katalogu Unity

Pro operátory SQL založené na Unity Catalogu, které volají externí API, použijte připojení HTTP ve službě Unity Catalog pro bezpečné ukládání přihlašovacích údajů:

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

Pak použijte připojení ve službě SQL UDF s funkcí http_request() . Podrobnosti najdete v tématu Připojení k externím službám HTTP.

WorkspaceClient

Pro operátory python-run-function můžete použít Azure Databricks WorkspaceClient pro přístup k prostředkům pracovního prostoru a externím rozhraním API:

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

Vytvořte kompletní uživatelem definovaný operátor python-run-function

Následující kroky vás provedou vytvořením operátoru python-run-function od začátku.

Krok 1: Definování logiky

Napište funkci run() v poznámkovém bloku:

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}

Krok 2: Testování funkce

Interaktivní testování funkce s ukázkovými daty:

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()

Krok 3: Vytvoření konfigurace YAML

Definujte metadata operátora, konfigurační pole a porty v souboru 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

Krok 4: Kombinování logiky a YAML

Přidejte pole run_function a ports vytvořte kompletní soubor YAML. Uložte ho do svého pracovního prostoru, například /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}

Krok 5: Registrace operátoru

Přidejte cestu k .user_defined_operators.yaml souboru:

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

Krok 6: Použití operátoru v Návrháři Lakeflow

Otevřete Lakeflow Designer a ověřte, že se operátor zobrazuje v paletě operátorů. Přetáhněte ho na plátno, připojte vstup, nakonfigurujte název sloupce a spusťte náhled.

Vytvoření kompletního uživatelem definovaného operátoru UC

Následující kroky vás provedou vytvořením operátoru UC založeného na uc-udf.

Krok 1: Definování logiky

Napište a otestujte logiku funkce v poznámkovém bloku:

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

Krok 2: Vytvoření konfigurace YAML

Definujte metadata operátora, konfigurační pole a porty:

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

Krok 3: Kombinování logiky a YAML

Vytvořte funkci Unity Catalog s yaML vloženou jako docstring:

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)
$$

Krok 4: Testování funkce

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

Krok 5: Registrace operátoru

Přidejte do .user_defined_operators.yaml souboru odkaz na funkci Katalogu Unity:

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

Krok 6: Použití operátoru v Návrháři Lakeflow

Otevřete Lakeflow Designer a ověřte, že se operátor zobrazuje v paletě operátorů. Přetáhněte ho na plátno, připojte vstup a spusťte náhled.

Troubleshooting

Issue	Řešení
Operátor se nezobrazuje v Návrháři Lakeflow.	Zkontrolujte, jestli `.user_defined_operators.yaml` existuje, a vypíše vaši funkci nebo cestu k souboru. U `python-run-function` operátorů ověřte cestu k souboru a že je soubor YAML přístupný.
Ověření schématu se nezdařilo.	Ověřte svůj YAML proti oficiálnímu schématu na adrese `https://your-workspace.cloud.databricks.com/static/schemas/user-defined-operator-v0.1.0.json`.
Přístup odepřen.	U operátorů založených na UC ověřte, že uživatelé mají `EXECUTE` k funkci a `USE SCHEMA` ke schématu. U `python-run-function` operátorů ověřte, že uživatelé mají k souboru YAML přístup pro čtení.
`python-run-function` operátor selže při běhu.	Zkontrolujte, zda signatura funkce `run()` odpovídá `def run(config, inputs, spark)`. Ověřte, že názvy portů v kódu odpovídají YAML a že návratové klíče slovníku odpovídají hodnotám výstupního portu `name` .
UDTF vrací nesprávné typy.	Návratové typy UDTF musí být explicitní; Nemůžete odkazovat na vstupní typy sloupců.

Povolení

Povolení	Purpose
Přístup pro čtení k `.user_defined_operators.yaml`.	Seznamte se s operátorem.
Přístup pro čtení k souboru YAML (`python-run-function` pouze).	Načtěte definici operátoru.
EXECUTE na funkci v Unity Catalogu (pouze pro operátory založené na UC).	Spusťte operátor.
USE SCHEMA na schématu (pouze operátory na bázi UC).	Přejděte ke schématu, ve kterém je funkce vytvořena.
Další oprávnění	V závislosti na vašem operátoru můžou uživatelé vyžadovat další oprávnění. Například `USE CONNECTION` v připojení ke katalogu Unity pro volání rozhraní HTTP API.

Další prostředky

Prozkoumejte následující kurzy:

Example	Typ	Description
Odesílatel e-mailu Gmailu	`python-run-function`	Odešle data datového rámce jako přílohu e-mailu CSV přes Gmail.
Kalkulačka složeného úroku	`uc-udf`	Výpočet budoucích investičních hodnot pomocí vzorce složeného úroku
Shlukování K-means	`uc-udtf`	Segmentace dat do clusterů pomocí knihovny scikit-learn
Odeslat zprávu Slack	`uc-udf`	Odesílání oznámení do kanálů Slack prostřednictvím rozhraní API
Všechny widgety uživatelského rozhraní	`uc-udf`	Referenční operátor zobrazuje všechny dostupné widgety uživatelského rozhraní.

Úplný přehled schématu YAML najdete v tématu Referenční příručka kódu YAML pro uživatelem definovaný operátor.

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-06-25

Uživatelem definované operátory v Návrháři Lakeflow

Jak fungují uživatelem definované operátory

Logika operátoru

Logika funkce run uživatelsky definovaného operátoru v Pythonu

Logika operátoru UDF a UDTF

Konfigurace YAML

Přístavy

YAML pro operátory funkce run v Pythonu

YAML pro funkce katalogu Unity

Zaregistrujte a nasaďte operátor do aplikace Lakeflow Designer

Aktualizace nebo odebrání operátoru

Pokročilé konfigurace

Režim náhledu

Připojení katalogu Unity

WorkspaceClient

Vytvořte kompletní uživatelem definovaný operátor python-run-function

Krok 1: Definování logiky

Krok 2: Testování funkce

Krok 3: Vytvoření konfigurace YAML

Krok 4: Kombinování logiky a YAML

Krok 5: Registrace operátoru

Krok 6: Použití operátoru v Návrháři Lakeflow

Vytvoření kompletního uživatelem definovaného operátoru UC

Krok 1: Definování logiky

Krok 2: Vytvoření konfigurace YAML

Krok 3: Kombinování logiky a YAML

Krok 4: Testování funkce

Krok 5: Registrace operátoru

Krok 6: Použití operátoru v Návrháři Lakeflow

Troubleshooting

Povolení

Další prostředky

Váš názor

Další materiály