Door de gebruiker gedefinieerde operators in Lakeflow Designer

Met Lakeflow Designer kunt u door de gebruiker gedefinieerde operators maken die rechtstreeks in het canvas naast ingebouwde operators worden weergegeven. Gebruik deze om Lakeflow Designer uit te breiden met uw eigen bedrijfslogica, berekeningen of integraties.

Er zijn drie typen door de gebruiker gedefinieerde operators:

  • python-run-function: een zelfstandig YAML-bestand met inline-Python opgeslagen in de werkruimte. Het meest geschikt voor transformaties op DataFrame-niveau en externe integraties. Machtigingen worden beheerd op het niveau van het werkruimtebestand.
  • uc-udf: Omsluit een scalaire functie van Unity Catalog. Het meest geschikt voor transformaties op kolomniveau. Toegang wordt beheerd door unity-catalogusmachtigingen.
  • uc-udtf: Omsluit een tabelwaardefunctie van Unity Catalog. Het meest geschikt voor transformaties op tabelniveau, zoals ML-clustering en -aggregatie. Toegang wordt beheerd door unity-catalogusmachtigingen.
Feature python-run-function uc-udf uc-udtf
Voorbeeld van een toepassing DataFrame-transformaties, API-integraties, e-mailmeldingen Berekeningen op kolomniveau (BMI, rente) ML-clustering, aggregatie tussen rijen
Invoer Dataframes Enkele waarden Hele tabel, rij per rij
Uitvoer Dataframes Enkele waarde Tabel (meerdere rijen)
Vereist Unity Catalog-functie No Yes Yes
Toegangsbeheer Machtigingen voor werkruimtebestanden Machtigingen voor Unity Catalog (EXECUTE, USE SCHEMA) Machtigingen voor Unity Catalog (EXECUTE, USE SCHEMA)
Ondersteunde talen Alleen Python SQL of Python in een SQL-wrapper SQL of Python in een SQL-wrapper

Hoe door de gebruiker gedefinieerde operators werken

Een door de gebruiker gedefinieerde operator bestaat uit:

  • Operatorlogica: De code die wordt uitgevoerd wanneer de operator wordt uitgevoerd. Dit kan een inline Python run() functie (voor python-run-function) of een Unity Catalog-functie (voor uc-udf en uc-udtf).
  • YAML-configuratie: Vertelt Lakeflow Designer hoe de operator in de gebruikersinterface kan worden weergegeven, waaronder de naam, beschrijving, invoerparameters, UI-widgets en poorten van de operator. Alle operatortypen maken gebruik van het user-defined-operator-v0.1.0 schema.
  • Registratiebestand: een vermelding in .user_defined_operators.yaml waarmee Lakeflow Designer de operator kan vinden.

Operatorlogica

Python-run-functie voor door de gebruiker gedefinieerde operatorlogica

Elke python-run-function operator moet een run() functie definiëren:

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config: door de gebruiker geconfigureerde waarden uit de gebruikersinterface, gesleuteld op eigenschapsnaam.
  • inputs: Invoerdataframes, gesleuteld door invoerpoort name.
  • spark: De actieve SparkSession.
  • Retourneert: Een woordenlijst die uitvoerpoortwaarden name aan DataFrames toetoewijzingt.

In het volgende voorbeeld worden rijen uit een invoerdataframe gefilterd:

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

Als uw operator externe PIP-pakketten vereist, voegt u het environment veld toe aan de YAML:

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

Operatorlogica voor UDF en UDTF

U kunt UC-functies schrijven in SQL of Python. Python functies worden verpakt in een SQL CREATE FUNCTION-instructie:

SQL-functie:

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-functie (verpakt in 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's verwerken één waarde tegelijk en retourneren een berekende waarde. UDTF's verwerken tabellen rij voor rij en kunnen over alle rijen heen een toestand behouden. Gebruiken uc-udf voor transformaties op kolomniveau en uc-udtf voor bewerkingen zoals ML-clustering of aggregatie.

Daarnaast moet u voor UDTF's drie belangrijke methoden definiëren: __init__(), eval() en 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-retourtabellen moeten vaste, expliciete typen hebben. U kunt niet verwijzen naar invoerkolomtypen in de retourconfiguratie.

YAML-configuratie

De YAML-configuratie vertelt Lakeflow Designer hoe de operator in de gebruikersinterface kan worden weergegeven. Hiermee definieert u de naam, beschrijving, invoerparameters, UI-widgets en poorten van de operator. Elk configuratieveld is een eigenschap met een type, titel en optionele x-ui widgethints:

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

Zie de YAML-referentie voor door de gebruiker gedefinieerde operator voor meer informatie over het YAML-schema, inclusief alle widgettypen en configuratieopties.

Ports

Poorten definiëren de invoer en uitvoer voor uw operator:

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

YAML voor Python functieoperators uitvoeren

Voor python-run-function-operators is het YAML-bestand zelfstandig en bevat het veld run_function met inline-Python-code:

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 voor Unity Catalog-functies

Sluit voor UC-operators de YAML-configuratie in als opmerking of docstring in uw functie.

In SQL (gebruik /* ... */ opmerking):

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 (gebruik """ ... """ 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)
$$;

Uw operator registreren en uitrollen in Lakeflow Designer

Registreer je operator in een .user_defined_operators.yaml-bestand zodat deze in Lakeflow Designer wordt weergegeven:

  • Werkruimteniveau: Plaats het bestand in de hoofdmap van uw werkruimte om de operator zichtbaar te maken voor alle gebruikers.
  • Gebruikersniveau: Plaats het bestand in de basismap van de gebruiker (/Workspace/Users/<user-name>/.user_defined_operators.yaml) om operators alleen zichtbaar te maken voor u.

De operators: sectie ondersteunt bestandspaden, functieverwijzingen van Unity Catalog en glob-patronen. U kunt invoertypen combineren:

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

Een operator bijwerken of verwijderen

Wanneer u de code van een operator wijzigt, vernieuwt u de door de gebruiker gedefinieerde operators om de wijziging te laden. Klik op het tabblad Operators van het menu op het pictogram Vernieuwen..

  • Als de operator hetzelfde version behoudt, wordt bij het vernieuwen de bijgewerkte code geladen.
  • Als de operator een nieuwe versie van version heeft, vraagt de operator op het canvas u na het vernieuwen om hiernaar te upgraden (of de huidige versie te behouden).

Als u een operator uit Lakeflow Designer wilt verwijderen, verwijdert u de vermelding uit .user_defined_operators.yaml. Voor de uc-udf- en uc-udtf-operators kunt u de onderliggende Unity Catalog-functie ook met DROP FUNCTION laten vallen als u die niet langer nodig hebt.

Geavanceerde configuraties

Preview-modus

Lakeflow Designer ondersteunt previews in de ontwerpmodus. Voor operators die externe API's aanroepen of schrijven naar externe systemen, voegt u een is_preview configuratie-eigenschap toe, zodat u neveneffecten tijdens de preview-versie kunt overslaan. Wanneer de preview-modus is ingeschakeld, moeten gebruikers expliciet op Uitvoeren klikken om de operator met bijwerkingen uit te voeren.

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

Lakeflow Designer stelt deze waarde true automatisch in op tijdens de preview. Vink dit in uw logica aan om neveneffecten over te slaan:

# 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-verbindingen

Voor OP UC gebaseerde SQL-operators die externe API's aanroepen, gebruikt u HTTP-verbindingen van Unity Catalog om veilig referenties op te slaan:

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

Gebruik vervolgens de verbinding in uw SQL UDF met de http_request() functie. Zie Verbinding maken met externe HTTP-services voor meer informatie.

WorkspaceClient

Voor python-run-function-operators kunt u de Azure Databricks WorkspaceClient gebruiken voor toegang tot werkruimtebronnen en externe API's:

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

Een volledige door de gebruiker gedefinieerde python-run-function-operator maken

In de volgende stappen doorloopt u hoe u helemaal vanaf nul een python-run-function-operator maakt.

Stap 1: De logica definiëren

Schrijf uw run() functie in een notebook:

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}

Stap 2: De functie testen

Test de functie interactief met voorbeeldgegevens:

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

Stap 3: de YAML-configuratie maken

Definieer de metagegevens van de operator, configuratievelden en poorten in een YAML-bestand:

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

Stap 4: Logica en YAML combineren

Voeg de velden run_function en ports toe om het volledige YAML-bestand te maken. Sla deze op in uw werkruimte, bijvoorbeeld /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}

Stap 5: De operator registreren

Voeg het bestandspad toe aan uw .user_defined_operators.yaml bestand:

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

Stap 6: De operator gebruiken in Lakeflow Designer

Open Lakeflow Designer en controleer of de operator wordt weergegeven in het operatorpalet. Sleep deze naar het canvas, verbind een invoer, configureer de kolomnaam en voer een voorbeeld uit.

Een volledige door de gebruiker gedefinieerde UC-operator maken

De volgende stappen begeleiden u bij het maken van een op UC gebaseerde uc-udf-operator.

Stap 1: De logica definiëren

Uw functielogica schrijven en testen in een notebook:

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

Stap 2: de YAML-configuratie maken

Definieer de metagegevens van de operator, configuratievelden en poorten:

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

Stap 3: De logica en YAML combineren

Maak de Unity Catalog-functie met de YAML ingesloten als 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)
$$

Stap 4: De functie testen

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

Stap 5: De operator registreren

Voeg de Unity Catalog-functiereferentie toe aan uw .user_defined_operators.yaml-bestand:

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

Stap 6: De operator gebruiken in Lakeflow Designer

Open Lakeflow Designer en controleer of de operator wordt weergegeven in het operatorpalet. Sleep het naar het canvas, verbind een invoer en voer een voorbeeld uit.

Troubleshooting

Issue Solution
De operator wordt niet weergegeven in Lakeflow Designer. Controleer of deze .user_defined_operators.yaml bestaat en vermeldt uw functie of bestandspad. Voor python-run-function-operators: controleer het bestandspad en of het YAML-bestand toegankelijk is.
Schemavalidatie mislukt. Controleer uw YAML op basis van het officiële schema op https://your-workspace.cloud.databricks.com/static/schemas/user-defined-operator-v0.1.0.json.
De machtiging is geweigerd. Controleer voor op UC gebaseerde operators of gebruikers EXECUTE op de functie en USE SCHEMA op het schema hebben. Controleer voor python-run-function operators dat gebruikers leestoegang hebben tot het YAML-bestand.
python-run-function de operator mislukt tijdens runtime. Controleer of de run() functiehandtekening overeenkomt def run(config, inputs, spark)met . Controleer of poortnamen in de code overeenkomen met de YAML en of de retourwoordenlijstsleutels overeenkomen met uitvoerpoortwaarden name .
UDTF retourneert verkeerde typen. UDTF-retourtypen moeten expliciet zijn; u kunt niet verwijzen naar invoerkolomtypen.

Permissions

Toestemming Purpose
Leestoegang tot .user_defined_operators.yaml. Ontdek de operator.
Leestoegang tot het YAML-bestand (python-run-function alleen). Laad de operatordefinitie.
EXECUTE op de Unity Catalog-functie (alleen op UC gebaseerde operators). Voer de operator uit.
USE SCHEMA in het schema (alleen UC-gebaseerde operatoren). Open het schema waarin de functie is gemaakt.
Andere machtigingen Afhankelijk van uw operator hebben gebruikers mogelijk andere machtigingen nodig. Bijvoorbeeld USE CONNECTION op een Unity Catalog-verbinding voor HTTP-API-aanroepen.

Aanvullende bronnen

Bekijk de volgende zelfstudies:

Example Typ Description
Afzender van Gmail-e-mail python-run-function DataFrame-gegevens verzenden als een CSV-e-mailbijlage via Gmail.
Rekenmachine voor samengestelde rente uc-udf Bereken toekomstige investeringswaarden met behulp van de formule voor samengestelde rente.
K-means-clustering uc-udtf Segmenteer gegevens in clusters met scikit-learn.
Slack-bericht verzenden uc-udf Meldingen verzenden naar Slack-kanalen via API.
Alle UI-widgets uc-udf Verwijzingsoperator die alle beschikbare UI-widgets weergeeft.

Zie de YAML-verwijzing voor door de gebruiker gedefinieerde operatoren voor een volledige verwijzing naar het YAML-schema.

  • Last updated on