Schreibe Logik für benutzerdefinierte Operatoren von Lakeflow Designer

Die Operatorlogik ist der Code, der von einem benutzerdefinierten Operator ausgeführt wird. Für einen python-run-function Operator handelt es sich um eine Inline-Python run() Funktion. Für einen uc-udf oder uc-udtf Operator handelt es sich um eine Unity-Catalog-Funktion, die in SQL oder Python geschrieben ist.

Das Schreiben der Logik ist ein Schritt beim Aufbau eines freigegebenen benutzerdefinierten Operators. Nachdem du die Logik geschrieben und getestet hast, beschreibst du den Operator in YAML, sodass Lakeflow Designer weiß, wie er ihn in der Benutzeroberfläche anzeigen kann, und registrierst du ihn dann, damit er im Operatoren-Bereich erscheint. Um alle Schritte für einen Operator zusammen zu sehen, folgen Sie einem Tutorial wie Tutorial: Operator für Zinseszinsrechner.

Schreibe einen Python-Run-Function-Operator

Jeder python-run-function Operator muss eine run() Funktion definieren:

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config: Vom Benutzer konfigurierte Werte aus der Benutzeroberfläche, schlüsselt nach Eigenschaftsname.
  • inputs: Eingabe-DataFrames, nach Eingabeport name.
  • spark: Die aktive SparkSession.
  • Ergebnis: Ein Wörterbuch, das name-Ausgabeportwerte DataFrames zuordnet.

Im folgenden Beispiel werden Zeilen aus einem Eingabedatenframe gefiltert:

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

Um die Funktion vor der Registrierung zu testen, rufen Sie sie in einem Notizbuch mit Beispieldaten auf:

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

Wenn Ihr Operator externe Pip-Pakete benötigt, fügen Sie das environment Feld zum YAML hinzu:

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

Schreiben Sie einen UDF- oder UDTF-Operator

Du kannst Unity Catalog-Funktionen in SQL oder Python schreiben. Python Funktionen werden in eine SQL-CREATE FUNCTION-Anweisung eingeschlossen:

SQL-Funktion:

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-Funktion (in SQL eingeschlossen):

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

UDFs verarbeiten jeweils einen einzelnen Wert und geben einen berechneten Wert zurück. UDTFs verarbeiten Tabellenzeile nach Zeile und können den Zustand über alle Zeilen hinweg verwalten. Verwenden Sie uc-udf für Transformationen auf Spaltenebene und uc-udtf für Operationen wie ML-Clustering oder Aggregation.

Eine UDTF-Handler-Klasse definiert eval(), die für jede Eingabezeile ausgeführt wird. Optional definiert dies auch __init__(), um den Zustand einzurichten und terminate(), um nach allen Zeilen Endergebnisse zu liefern. Siehe Python benutzerdefinierte Tabellenfunktionen (UDTFs) im Unity-Katalog.

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.
        ...

Hinweis

UDTF-Rückgabetabellen müssen feste, explizite Typen aufweisen. Sie können in der Rückgabekonfiguration nicht auf Eingabespaltentypen verweisen.

Seiteneffekte während der Vorschau überspringen

Lakeflow Designer unterstützt Vorschauen im Entwurfsmodus. Fügen Sie für Operatoren, die externe APIs aufrufen oder in externe Systeme schreiben, eine is_preview Konfigurationseigenschaft hinzu, damit Sie während der Vorschau Nebenwirkungen überspringen können. Benutzer klicken auf Ausführen, um den Operator mit seinen Nebenwirkungen auszuführen. Der eingebaute Python-Operator benötigt diese Eigenschaft nicht, da er config["is_preview"] automatisch liest. Siehe Python-Vorschaumodus.

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

Lakeflow Designer setzt diesen Wert während der Vorschau automatisch auf true. Berücksichtigen Sie dies in Ihrer Logik, um Seiteneffekte zu vermeiden:

# 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

Rufen Sie externe APIs mit Unity Catalog HTTP-Verbindungen auf

Für benutzerdefinierte Operatoren, die auf SQL Unity Catalog-Funktionen basieren und externe APIs aufrufen, verwenden Sie HTTP-Verbindungen im Unity Catalog, um Zugangsdaten sicher zu speichern:

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

Verwenden Sie dann die Verbindung in Ihrer SQL-UDF mit der http_request() Funktion. Ausführliche Informationen finden Sie unter Herstellen einer Verbindung mit externen HTTP-Diensten.

Greifen Sie mit WorkspaceClient auf Arbeitsbereichsressourcen zu

Für python-run-function-Operatoren können Sie die Azure Databricks WorkspaceClient verwenden, um auf Arbeitsbereichsressourcen und externe APIs zuzugreifen:

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

Zusätzliche Ressourcen

  • Zuletzt aktualisiert um