Scrivere la logica per gli operatori definiti dall'utente di Lakeflow Designer

La logica dell'operatore è il codice che viene eseguito quando un operatore definito dall'utente viene eseguito. Per un python-run-function operatore, è una funzione run()Python inline. Per un operatore uc-udf o uc-udtf, si tratta di una funzione di Unity Catalog scritta in SQL o Python.

Scrivere la logica è un passaggio nella costruzione di un operatore condiviso definito dall'utente. Dopo aver scritto e testato la logica, descrivi l'operatore in YAML così Lakeflow Designer sa come mostrarlo nell'interfaccia utente, poi lo registri in modo che appaia nel pannello Operators. Per vedere tutti i passaggi insieme per un operatore, segui un tutorial come Tutorial: Operatore calcolatore di interessi composti.

Scrivi un operatore python-run-function

Ogni python-run-function operatore deve definire una run() funzione:

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config: valori configurati dall'utente nell'interfaccia utente, indicizzati per nome della proprietà.
  • inputs: DataFrame di input, indicizzati per porta di input name.
  • spark: La SparkSession attiva.
  • Valori restituiti: un dizionario che associa i valori delle porte di output name ai DataFrame.

L'esempio seguente filtra le righe da un dataframe di input:

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

Per testare la funzione prima di registrarla, chiamala in un notebook con dati di esempio:

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

Se l'operatore richiede pacchetti pip esterni, aggiungere il environment campo a YAML:

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

Scrivi un operatore UDF o UDTF

Puoi scrivere funzioni di Unity Catalog in SQL o Python. Le funzioni Python sono racchiuse in un'istruzione SQL CREATE FUNCTION:

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

Funzione Python (incapsulata 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)
$$;

Le funzioni definite dall'utente elaborano un singolo valore alla volta e restituiscono un valore calcolato. Le UDTF elaborano le tabelle riga per riga e possono mantenere uno stato condiviso tra tutte le righe. Utilizzare uc-udf per le trasformazioni a livello di colonna e uc-udtf per operazioni come il clustering ML o l'aggregazione.

Una classe handler UDTF definisce eval(), che viene eseguita per ogni riga di input. Opzionalmente, definisce anche __init__() per impostare lo stato e terminate() per ottenere i risultati finali dopo tutte le righe. Vedere Funzioni di tabella definite dall'utente di Python (UDTF) in Unity Catalog.

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

Annotazioni

Le tabelle restituite UDTF devono avere tipi fissi ed espliciti. Non è possibile fare riferimento ai tipi di colonna di input nella configurazione restituita.

Saltare gli effetti secondari durante le anteprime

Lakeflow Designer supporta le anteprime in modalità progettazione. Per gli operatori che chiamano API esterne o scrivono in sistemi esterni, aggiungere una is_preview proprietà config in modo da ignorare gli effetti collaterali durante l'anteprima. Gli utenti cliccano su Esegui per eseguire l'operatore con i relativi effetti collaterali. L'operatore Python integrato non ha bisogno di questa proprietà, perché legge config["is_preview"] automaticamente. Vedi modalità di anteprima Python.

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

Lakeflow Designer imposta automaticamente questo valore su true durante l'anteprima. Gestiscilo nella tua logica per ignorare gli effetti collaterali:

# 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

Chiama API esterne con connessioni HTTP di Unity Catalog

Per operatori definiti dall'utente basati su funzioni SQL Unity Catalog che chiamano API esterne, usa le connessioni HTTP Unity Catalog per memorizzare in modo sicuro le credenziali:

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

Usa quindi la connessione nella tua funzione SQL definita dall'utente con la funzione http_request(). Per informazioni dettagliate, vedere Connettersi a servizi HTTP esterni.

Accedi alle risorse dello spazio di lavoro con WorkspaceClient

Per gli operatori python-run-function, è possibile usare il Azure Databricks WorkspaceClient per accedere alle risorse dell'area di lavoro e alle API esterne:

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

Risorse aggiuntive

  • Ultimo aggiornamento: