Napisz logikę dla niestandardowych operatorów Lakeflow Designer

Logika operatora to kod wykonywany przez niestandardowy operator. Dla operatora python-run-function jest to funkcja inline w Pythonie run(). Dla operatora uc-udf lub uc-udtf jest to funkcja Unity Catalog napisana w języku SQL lub Python.

Napisanie logiki to jeden z etapów budowania współdzielonego operatora niestandardowego. Po napisaniu i przetestowaniu logiki opisujesz operator w YAML , aby Lakeflow Designer wiedział, jak go wyświetlić w interfejsie, a następnie rejestrujesz go tak, by pojawiał się w panelu operatorów . Aby zobaczyć wszystkie kroki razem dla jednego operatora, skorzystaj z poradnika, takiego jak Samouczek: operator kalkulatora procentu składanego.

Napisz operator python-run-function

Każdy python-run-function operator musi zdefiniować run() funkcję:

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config: wartości skonfigurowane przez użytkownika z interfejsu użytkownika, kluczowane według nazwy właściwości.
  • inputs: Wejściowe ramki danych, kluczowane przez port namewejściowy .
  • spark: aktywna platforma SparkSession.
  • Zwraca: Słownik przypisujący wartości portu wyjściowego name do ramek danych.

Poniższy przykład filtruje wiersze z wejściowej ramki danych:

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

Aby przetestować funkcję przed jej rejestracją, wywołaj ją w notatniku z przykładowymi danymi:

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

Jeśli operator wymaga zewnętrznych pakietów pip, dodaj pole environment do pliku YAML:

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

Napisz operator UDF lub UDTF

Możesz pisać funkcje Unity Catalog w SQL lub Python. Funkcje języka Python są ujęte w instrukcji SQL CREATE FUNCTION:

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

Funkcja Pythona (opakowana w 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)
$$;

Funkcje definiowane przez użytkownika przetwarzają po jednej wartości naraz i zwracają obliczoną wartość. Funkcje tabelaryczne zdefiniowane przez użytkownika (UDTF) przetwarzają tabele wiersz po wierszu i mogą utrzymywać stan między wszystkimi wierszami. Użyj uc-udf do przekształceń na poziomie kolumn, a uc-udtf do operacji, takich jak grupowanie metodą uczenia maszynowego lub agregacja.

Klasa obsługi UDTF definiuje eval(), które jest uruchamiane dla każdego wiersza wejściowego. Opcjonalnie definiuje także __init__() do ustawiania stanu oraz terminate() do zwracania końcowych wyników po wszystkich wierszach. Zobacz Funkcje tabeli definiowane przez użytkownika (UDTF) w języku Python w katalogu Unity.

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

Uwaga / Notatka

Tabele zwracane przez funkcję UDTF muszą mieć stałe, jawne typy. Nie można odwoływać się do typów kolumn wejściowych w konfiguracji zwracanej.

Pomiń efekty uboczne podczas podglądów

Program Lakeflow Designer obsługuje podglądy w trybie projektowania. W przypadku operatorów, które wywołują zewnętrzne interfejsy API lub zapisują do systemów zewnętrznych, dodaj właściwość is_preview, aby podczas podglądu można było pominąć efekty uboczne. Użytkownicy klikają Run, aby uruchomić operator wraz z jego efektami ubocznymi. Wbudowany operator Python nie potrzebuje tej właściwości, ponieważ automatycznie odczytuje config["is_preview"]. Zobacz tryb podglądu języka Python.

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

Program Lakeflow Designer automatycznie ustawia tę wartość na true w trakcie korzystania z wersji zapoznawczej. Uwzględnij to w swojej logice, aby pomijać efekty uboczne:

# 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

Wywołuj zewnętrzne API z połączeniami HTTP w Unity Catalog

Dla niestandardowych operatorów opartych na funkcjach SQL Unity Catalog, które wywołują zewnętrzne API, użyj połączeń HTTP Unity Catalog do bezpiecznego przechowywania danych uwierzytelniających:

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

Następnie użyj połączenia w swojej funkcji UDF SQL z funkcją http_request(). Aby uzyskać szczegółowe informacje, zobacz Nawiązywanie połączenia z zewnętrznymi usługami HTTP.

Uzyskaj dostęp do zasobów przestrzeni roboczej za pomocą WorkspaceClient

W przypadku operatorów python-run-function można użyć Azure Databricks WorkspaceClient, aby uzyskać dostęp do zasobów obszaru roboczego i zewnętrznych interfejsów API:

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

Dodatkowe zasoby

  • Ostatnia aktualizacja o