Operadores definidos pelo utilizador no Lakeflow Designer

O Lakeflow Designer permite-lhe criar operadores definidos pelo utilizador que aparecem diretamente na tela ao lado de operadores incorporados. Use-os para expandir o Lakeflow Designer com a sua própria lógica de negócio, cálculos ou integrações.

Existem três tipos de operadores definidos pelo utilizador:

• python-run-function: Um ficheiro YAML independente com Python incorporado armazenado no espaço de trabalho. Ideal para transformações ao nível do DataFrame e integrações externas. As permissões são geridas ao nível do ficheiro do espaço de trabalho.
• uc-udf: Encapsula uma função escalar do Unity Catalog. Ideal para transformações ao nível da coluna. O acesso é regido pelas permissões do Catálogo Unity.
• uc-udtf: Encapsula uma função com valor de tabela do Unity Catalog. Ideal para transformações ao nível da tabela, como agrupamento por aprendizagem automática e agregação. O acesso é regido pelas permissões do Catálogo Unity.

Como funcionam os operadores definidos pelo utilizador

Um operador definido pelo utilizador consiste em:

• Lógica do operador: O código que corre quando o operador é executado. Isto pode ser uma função inline Python run() (para python-run-function) ou uma função Unity Catalog (para uc-udf e uc-udtf).
• Configuração YAML: Indica ao Lakeflow Designer como apresentar o operador na interface, incluindo o nome do operador, descrição, parâmetros de entrada, widgets UI e portas. Todos os tipos de operadores utilizam o user-defined-operator-v0.1.0 esquema.
• Ficheiro de registo: Uma entrada .user_defined_operators.yaml que permite ao Lakeflow Designer descobrir o operador.

Lógica de operadores

Python função de execução lógica de operadores definida pelo utilizador

Cada python-run-function operador deve definir uma run() função:

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

• config: Valores configurados pelo utilizador na interface, identificados pelo nome da propriedade.
• inputs: DataFrames de entrada, codificados pela porta namede entrada .
• spark: A SparkSession em utilização.
• Retornos: Um dicionário que mapeia valores de porta name de saída para DataFrames.

O exemplo seguinte filtra linhas de um DataFrame de entrada:

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

Se o seu operador precisar de pacotes de pips externos, adicione o environment campo ao YAML:

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

Lógica do operador UDF e UDTF

Podes escrever funções UC em SQL ou Python. As funções Python são encapsuladas numa instrução SQL CREATE FUNCTION:

Função 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);

Função Python (encapsulada em 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)
$$;

Os UDFs processam um único valor de cada vez e devolvem um valor calculado. Os UDTFs processam tabelas linha a linha e podem manter o estado em todas as linhas. Use uc-udf para transformações ao nível de coluna e uc-udtf para operações como clustering ou agregação de ML.

Além disso, os UDTFs exigem que defina três métodos-chave: __init__(), eval(), e 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.

Configuração YAML

A configuração YAML indica ao Lakeflow Designer como apresentar o operador na interface. Define o nome do operador, descrição, parâmetros de entrada, widgets de interface e portas. Cada campo de configuração é uma propriedade com um tipo, título e dicas opcionais x-ui de widget:

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

Para detalhes completos sobre o esquema YAML, incluindo todos os tipos de widgets e opções de configuração, veja User-defined operator YAML reference.

Portas

As portas definem as entradas e saídas do seu operador:

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

YAML para operadores da função de execução em Python

Para operadores python-run-function, o ficheiro YAML é autónomo e inclui um campo run_function com código de Python inline:

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 para funções do Catálogo Unity

Para operadores baseados em UC, incorpore a configuração YAML como comentário ou docstring na sua função.

Em SQL (usar /* ... */ comentário):

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

Em Python (usar """ ... """ 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)
$$;

Registe e envia o teu operador para o Lakeflow Designer

Para o seu operador aparecer no Lakeflow Designer, registe-o num .user_defined_operators.yaml ficheiro:

• Nível do espaço de trabalho: Coloque o ficheiro na raiz do seu espaço de trabalho para tornar o operador visível para todos os utilizadores.
• Nível de utilizador: Coloque o ficheiro na pasta inicial do utilizador (/Workspace/Users/<user-name>/.user_defined_operators.yaml) para tornar os operadores visíveis apenas para si.

A secção operators: suporta caminhos de ficheiros, referências a funções do Unity Catalog e padrões glob. Pode misturar os tipos de entrada:

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

Atualizar ou remover um operador

Quando alterar o código de um operador, atualize os operadores definidos pelo utilizador para carregar a alteração. No separador Operadores do menu, clique .

• Se o operador mantiver o mesmo version, a atualização carrega o código atualizado.
• Se o operador tiver um novo version, o operador na área de desenho solicita-te que atualizes para essa versão (ou que mantenhas a versão atual) depois de recarregares a página.

Para remover um operador do Lakeflow Designer, elimine a sua entrada de .user_defined_operators.yaml. Para os operadores uc-udf e uc-udtf, também pode eliminar a função subjacente do Unity Catalog com DROP FUNCTION, caso já não precise dela.

Configurações avançadas

Modo de pré-visualização

O Lakeflow Designer suporta pré-visualizações enquanto está em modo de design. Para operadores que chamam APIs externas ou escrevem em sistemas externos, adicione uma is_preview propriedade de configuração para poder saltar efeitos secundários durante a pré-visualização. Quando o modo de pré-visualização está ativado, os utilizadores precisam de clicar explicitamente em Executar para executar o operador com efeitos secundários.

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

O Lakeflow Designer define automaticamente este valor para true durante a pré-visualização. Verifica isso na tua lógica para evitar efeitos secundários:

# 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

Conexões do Unity Catalog

Para operadores SQL baseados em UC que chamam APIs externas, utilize ligações HTTP do Unity Catalog para armazenar credenciais de forma segura:

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

Em seguida, utilize a ligação na sua SQL UDF com a função http_request(). Para mais detalhes, consulte Ligar a serviços HTTP externos.

WorkspaceClient

Para operadores python-run-function, pode usar o Azure Databricks WorkspaceClient para aceder a recursos de espaço de trabalho e APIs externas:

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

Criar um operador definido pelo utilizador com função de execução completa em python

Os passos seguintes orientam na criação de um operador python-run-function a partir do zero.

Passo 1: Definir a lógica

Escreva a sua run() função num caderno:

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}

Passo 2: Testar a função

Teste a função de forma interativa com dados de exemplo:

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

Passo 3: Criar a configuração YAML

Defina os metadados do operador, campos de configuração e portas num ficheiro 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

Passo 4: Combinar a lógica e o YAML

Adicione os run_function campos e ports para criar o ficheiro YAML completo. Guarde-o no seu espaço de trabalho, por /Workspace/Users/<user-name>/udos/add_timestamp.yamlexemplo:

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}

Passo 5: Registar o operador

Adicione o caminho do ficheiro ao seu .user_defined_operators.yaml ficheiro:

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

Passo 6: Usar o operador no Lakeflow Designer

Abra o Lakeflow Designer e verifique se o operador aparece na paleta do operador. Arrasta-o para a área de desenho, liga-o a uma entrada, configura o nome da coluna e executa a pré-visualização.

Criar um operador UC completo definido pelo utilizador

Os passos seguintes explicam como criar um operador baseado uc-udf na UC.

Passo 1: Definir a lógica

Escreve e testa a tua lógica funcional num caderno:

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

Passo 2: Criar a configuração YAML

Defina os metadados do operador, campos de configuração e portas:

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

Passo 3: Combinar a lógica e o YAML

Crie a função Unity Catalog com o YAML incorporado como 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)
$$

Passo 4: Testar a função

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

Passo 5: Registar o operador

Adicione a referência da função Unity Catalog ao seu .user_defined_operators.yaml ficheiro:

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

Passo 6: Usar o operador no Lakeflow Designer

Abra o Lakeflow Designer e verifique se o operador aparece na paleta do operador. Arrasta-o para a tela, liga uma entrada e executa uma pré-visualização.

Troubleshooting

Permissions

Recursos adicionais

Explore os seguintes tutoriais:

Para uma referência completa ao esquema YAML, veja Referência YAML do operador definido pelo utilizador.