Escrever lógica para operadores personalizados do Lakeflow Designer

A lógica do operador é o código que um operador personalizado executa. Para um python-run-function operador, é uma função Python run() inline. Para um operador uc-udf ou uc-udtf, trata-se de uma função do Catálogo do Unity escrita em SQL ou Python.

Escrever a lógica é um passo na construção de um operador personalizado compartilhado. Depois de escrever e testar a lógica, você descreve o operador em YAML para que o Lakeflow Designer saiba como exibi-lo na interface, e então o registra para que apareça no painel de Operadores. Para ver todos os passos juntos para um operador, siga um tutorial como Tutorial: Operador de calculadora de juros compostos.

Escreva um operador python-run-function

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}

Para testar a função antes de registrá-la, chame-a em um notebook com dados de exemplo:

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

Escreva um operador UDF ou UDTF

Você pode escrever funções do Catálogo Unity em 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.

Uma classe manipuladora UDTF define eval(), que é executada para cada linha de entrada. Opcionalmente, também define __init__() para configurar o estado e terminate() para gerar os resultados finais após todas as linhas. Consulte funções de tabela definidas pelo usuário (UDTFs) em Python no 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.
        ...

Observação

As tabelas de retorno UDTF devem ter tipos fixos e explícitos. Você não pode referenciar tipos de coluna de entrada na configuração de retorno.

Ignorar efeitos colaterais durante prévias

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. Os usuários clicam em Executar para rodar o operador com seus efeitos colaterais. O operador Python embutido não precisa dessa propriedade, porque ele lê config["is_preview"] automaticamente. Veja o modo de pré-visualização em Python.

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

Chame APIs externas com conexões HTTP do Unity Catalog

Para operadores personalizados baseados em funções SQL Unity Catalog que chamam APIs externas, use conexõ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, 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.

Acesse recursos de workspace com o 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

Recursos adicionais

  • Última atualização em