Lógica para escrever 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 Unity Catalog escrita em SQL ou Python.

Escrever a lógica é um passo na construção de um operador personalizado partilhado. Depois de escrever e testar a lógica, descreve o operador em YAML para que o Lakeflow Designer saiba como o mostrar na interface, e depois regista-o 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.

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

Para testar a função antes de a registar, chame-a num caderno 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 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

Escrever um operador UDF ou UDTF

Podes escrever funções do Unity Catalog 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.

Uma classe de manipulador UDTF define eval(), o qual é executado para cada linha de entrada. Opcionalmente, também define __init__() para configurar o estado e terminate() para devolver os resultados finais após todas as linhas. Consulte Funções de tabela definidas pelo usuário (UDTFs) do 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.
        ...

Note

As tabelas de retorno do UDTF devem ter tipos fixos e explícitos. Não podes referenciar os tipos de colunas de entrada na configuração de retorno.

Ignorar os efeitos secundários durante as pré-visualizações

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. Os utilizadores clicam em Executar para executar o operador com os seus efeitos secundários. O operador Python incorporado não precisa desta propriedade, porque 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

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

Chame APIs externas com ligações HTTP do Unity Catalog

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

Aceda aos recursos do espaço de trabalho com o 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

Recursos adicionais

  • Última atualização a