Pular para o conteúdo principal
Última atualização em

Escrever lógica para operadores definidos pelo usuário do Lakeflow Designer

A lógica do operador é o código em execução quando ocorre a execução de um operador definido pelo usuário. Para um operador python-run-function, é uma função run() do Python inline. Para um operador uc-udf ou uc-udtf, é uma função do Unity Catalog escrita em SQL ou Python.

Escrever a lógica é o passo na criação de um operador compartilhado definido pelo usuário. Depois de escrever e testar a lógica, você descreve o operador em YAML para que o Lakeflow Designer saiba como mostrá-lo na UI, e então registro para que ele apareça no painel Operadores . Para ver todos os passos de um operador, siga um tutorial como Tutorial: Operador de calculadora de juros compostos.

Escrever um operador de função de execução Python​

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

Python
def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config : Valores configurados pelo usuário a partir da UI, mapeados por nome de propriedade.
  • inputs : DataFrames de entrada, chaveados por porta de entrada name.
  • spark : A SparkSession ativa.
  • Retorna : um dicionário mapeando os valores da porta de saída name para DataFrames.

O exemplo a seguir filtra linhas de um DataFrame de entrada:

Python
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 amostra:

Python
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 seu operador exigir pacotes pip externos, adicione o campo environment ao YAML:

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 Unity Catalog em SQL ou Python. As funções Python são envolvidas em uma instrução SQL CREATE FUNCTION:

Função SQL:

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 (agrupada em SQL):

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

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

Uma classe manipuladora UDTF define eval(), que é uma execução para cada linha de entrada. Opcionalmente, ele também define __init__() para configurar o estado e terminate() para produzir resultados finais após todas as linhas. Consulte Funções de tabela definidas pelo usuário (UDTFs) em Python no Unity Catalog.

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

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

Pular efeitos colaterais durante visualizações​

O Lakeflow Designer suporta pré-visualizações enquanto no modo de design. Para operadores que chamam APIs externas ou escrevem em sistemas externos, adicione uma propriedade de configuração is_preview para que você possa ignorar efeitos colaterais durante a pré-visualização. Os usuários clicam em execução para execução o operador com seus efeitos colaterais. O operador Python integrado não precisa desta propriedade, porque ele lê config["is_preview"] automaticamente. Consulte Modo de visualização do Python.

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

O Lakeflow Designer define automaticamente este valor como true durante a pré-visualização. Verifique isso em sua lógica para pular efeitos colaterais:

Python
# In a python-run-function
if config.get("is_preview"):
    return {"out": inputs["in"]}
SQL
-- In a UC function (SQL)
CASE WHEN is_preview THEN 'preview' ELSE /* actual work */ END

Chamar APIs externas com conexões HTTP do Unity Catalog​

Para operadores definidos pelo usuário com base em funções do SQL Unity Catalog que chamam APIs externas, use conexões HTTP do Unity Catalog para armazenar credenciais com segurança:

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

Acesse os recursos do workspace com WorkspaceClient​

Para os operadores python-run-function, você pode usar o Databricks WorkspaceClient para acessar recursos de workspace e APIs externas:

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

Outros recursos​

Nesta página