Operadores definidos pelo usuário no Lakeflow Designer

O Lakeflow Designer permite criar operadores definidos pelo usuário que aparecem diretamente na tela ao lado de operadores internos. Use-os para estender o Lakeflow Designer com sua própria lógica de negócios, cálculos ou integrações.

Você pode criar um operador local definido pelo usuário com o Código Genie para o arquivo atual, ou registrar um operador compartilhado definido pelo usuário para uso entre arquivos.

Operadores compartilhados definidos pelo usuário têm três tipos:

• python-run-function: um arquivo YAML independente com Python embutido armazenado no espaço de trabalho. Melhor para transformações no nível de DataFrame e integrações externas. As permissões são gerenciadas no nível do arquivo do workspace.
• uc-udf: encapsula uma função escalar do Unity Catalog. Ideal para transformações no nível da coluna. O acesso é regido pelas permissões do Catálogo do Unity.
• uc-udtf: encapsula uma função do Unity Catalog com valor de tabela. Melhor para transformações no nível da tabela, como agrupamento e agregação de aprendizado de máquina. O acesso é regido pelas permissões do Catálogo do Unity.

Crie um operador definido localmente pelo usuário com o Código Genie

Quando o Genie Code estiver disponível no seu espaço de trabalho, você pode criar um operador definido localmente pelo usuário a partir da tela do Lakeflow Designer:

1. No menu Operadores , selecione Criar operador definido pelo usuário.
2. Insira uma descrição e, opcionalmente, um nome. Selecione Gerar prévia.
3. Revise a configuração gerada e a função Python, depois selecione Aprovar e crie.

O Designer armazena a definição do operador local no arquivo atual e a adiciona ao menu Operadores . Use o menu de ações do operador para editar ou excluir a definição. Você deve remover todas as instâncias do operador da tela antes de poder excluir sua definição.

Como os operadores definidos pelo usuário funcionam

Um operador compartilhado definido pelo usuário consiste em:

• Lógica do operador: o código executado quando o operador é executado. Isso pode ser uma função Python run() embutida (para python-run-function) ou uma função do Catálogo do Unity (para uc-udf e uc-udtf).
• Configuração do YAML: informa ao Lakeflow Designer como apresentar o operador na interface do usuário, incluindo o nome do operador, a descrição, os parâmetros de entrada, os widgets de interface do usuário e as portas. Todos os tipos de operador usam o user-defined-operator-v0.1.0 esquema.
• Arquivo de registro: Uma entrada em .user_defined_operators.yaml que permite que o Lakeflow Designer descubra o operador.

Lógica do operador

Lógica do operador definido pelo usuário da função executada pelo Python

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 usuário na interface do usuário, codificados pelo nome da propriedade.
• inputs: DataFrames de entrada, codificado pela porta de entrada name.
• spark: o SparkSession ativo.
• Retornos: um dicionário que mapeia os valores da porta de saída name para DataFrames.

O exemplo a seguir 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 operador exigir pacotes pip externos, adicione o campo environment ao YAML:

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

Lógica do operador UDF e UDTF

Você pode escrever funções UC no SQL ou Python. Funções Python são envolvidas em uma 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)
$$;

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

Além disso, os UDTFs exigem que você defina três métodos principais: __init__()eeval()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 do YAML

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

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 obter detalhes completos sobre o esquema YAML, incluindo todos os tipos de widget e opções de configuração, consulte a referência YAML do operador definido pelo usuário.

Portas

As portas definem as entradas e saídas para o 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 de função de execução para Python

Para operadores python-run-function, o arquivo YAML é autônomo e inclui um campo run_function com código Python embutido:

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

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

Em SQL (use o 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 (use """ ... """ 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)
$$;

Registrar e implantar seu operador no Lakeflow Designer

Para que o operador apareça no Lakeflow Designer, registre-o em um .user_defined_operators.yaml arquivo:

• Nível do workspace: Coloque o arquivo na raiz do workspace para tornar o operador visível para todos os usuários.
• Nível de usuário: Coloque o arquivo na pasta inicial do usuário (/Workspace/Users/<user-name>/.user_defined_operators.yaml) para tornar os operadores visíveis somente para você.

A seção operators: dá suporte a caminhos de arquivo, referências de função do Unity Catalog e padrões glob. Você pode misturar 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

Ao alterar o código de um operador, atualize os operadores definidos pelo usuário para que a alteração seja carregada. Na guia Operadores do menu, clique no .

• Se o operador mantiver o mesmo version, a atualização carregará o código atualizado.
• Se o operador tiver uma nova version, o operador na tela solicitará que você faça upgrade (ou mantenha a versão atual) após a atualização.

Para remover um operador do Lakeflow Designer, exclua sua entrada de .user_defined_operators.yaml. Para operadores uc-udf e uc-udtf, você também pode excluir a função subjacente do Unity Catalog com DROP FUNCTION, caso não precise mais dela.

Configurações avançadas

Modo de visualização

O Lakeflow Designer dá suporte a visualizações enquanto estiver no modo de design. Para operadores que chamam APIs externas ou gravam em sistemas externos, adicione uma propriedade de configuração is_preview para poder ignorar efeitos colaterais durante a versão preliminar. Quando o modo de visualização está habilitado, os usuários precisam clicar explicitamente em Executar para executar o operador com efeitos colaterais.

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

Lakeflow Designer define automaticamente esse valor como true durante a visualização. Verifique isso na sua lógica para evitar efeitos colaterais:

# 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, use conexões HTTP do Catálogo do Unity para armazenar credenciais com segurança:

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, use a conexão na sua UDF SQL com a função http_request(). Para obter detalhes, consulte Conectar-se a serviços HTTP externos.

WorkspaceClient

Para operadores python-run-function, você pode usar o Azure Databricks WorkspaceClient para acessar recursos do 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 completo definido pelo usuário de python-run-function

As etapas a seguir mostram como criar um operador python-run-function do zero.

Etapa 1: Definir a lógica

Escreva sua run() função em um 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}

Etapa 2: Testar a função

Teste a função interativamente 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()

Etapa 3: Criar a configuração do YAML

Defina os metadados do operador, os campos de configuração e as portas em um arquivo 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

Etapa 4: Combinar a lógica e o YAML

Adicione os campos run_function e ports para criar o arquivo YAML completo. Salve isso no seu espaço de trabalho, por exemplo /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}

Etapa 5: Registrar o operador

Adicione o caminho do arquivo ao arquivo .user_defined_operators.yaml :

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

Etapa 6: Usar o operador no Lakeflow Designer

Abra o Lakeflow Designer e verifique se o operador aparece na paleta de operadores. Arraste-o para a tela, conecte uma entrada, configure o nome da coluna e execute uma versão preliminar.

Criar um operador completo definido pelo usuário da UC

As etapas a seguir mostram como criar um operador baseado em UC uc-udf.

Etapa 1: Definir a lógica

Escreva e teste sua lógica de função em um notebook:

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

Etapa 2: Criar a configuração do YAML

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

Etapa 3: Combinar a lógica e o YAML

Crie a função Unity Catalog com o YAML embutido como docstring. O exemplo cria a função em main.example_output. Crie o esquema primeiro se ele não existir:

CREATE SCHEMA IF NOT EXISTS main.example_output

CREATE OR REPLACE FUNCTION main.example_output.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)
$$

Etapa 4: Testar a função

Teste a função. Ela retorna 10.

SELECT main.example_output.double_value(5) AS result;

Etapa 5: Registrar o operador

Adicione a referência de função do Catálogo do Unity ao seu .user_defined_operators.yaml arquivo:

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

Etapa 6: Usar o operador no Lakeflow Designer

Abra o Lakeflow Designer e verifique se o operador aparece na paleta de operadores. Arraste-o para a tela, conecte uma entrada e execute uma visualização.

Troubleshooting

Permissions

Recursos adicionais

Explore os seguintes tutoriais:

Para obter uma referência completa ao esquema YAML, consulte a referência YAML do operador definido pelo usuário.