Lógica para implementar operadores definidos por el usuario de Lakeflow Designer

La lógica del operador es el código que se ejecuta cuando se ejecuta un operador definido por el usuario. Para un python-run-function operador, es una función Python run() en línea. Para un operador uc-udf o uc-udtf, es una función de Unity Catalog escrita en SQL o Python.

Escribir la lógica es un paso para construir un operador compartido definido por el usuario. Después de escribir y probar la lógica, describes el operador en YAML para que Lakeflow Designer sepa cómo mostrarlo en la interfaz de usuario, y luego lo registras para que aparezca en el panel de Operadores. Para ver todos los pasos juntos para un operador, sigue un tutorial como Tutorial: Operador de calculadora de interés compuesto.

Escribe un operador python-run-function

Cada python-run-function operador debe definir una run() función:

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config: Valores configurados por el usuario en la interfaz de usuario, indexados por nombre de propiedad.
  • inputs: DataFrames de entrada, indexados por el puerto de entrada name.
  • spark: La SparkSession activa.
  • Devuelve: Un diccionario que asigna los valores del puerto de salida name a DataFrames.

En el ejemplo siguiente se filtran las filas de un dataframe de entrada:

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

Para probar la función antes de registrarla, llámala en un cuaderno con datos de muestra:

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

Si el operador requiere paquetes pip externos, agregue el environment campo a YAML:

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

Escribe un operador UDF o UDTF

Puedes escribir funciones de Unity Catalog en SQL o Python. Las funciones de Python se envuelven en una instrucción SQL CREATE FUNCTION:

Función 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);

función Python (encapsulada en 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)
$$;

Las UDF procesan un único valor a la vez y devuelven un valor calculado. Las UDTF procesan tablas fila por fila y pueden mantener un estado a lo largo de todas las filas. Se usa uc-udf para transformaciones de nivel de columna y uc-udtf para operaciones como la agrupación en clústeres de ML o la agregación.

Una clase manejadora UDTF define eval(), que se ejecuta para cada fila de entrada. Opcionalmente, también define __init__() para configurar el estado y terminate() para obtener resultados finales después de procesar todas las filas. Consulte Funciones de tabla definidas por el usuario (UDF) de Python en el catálogo de Unity.

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

Las tablas de retorno UDTF deben tener tipos fijos y explícitos. No se puede hacer referencia a los tipos de columna de entrada en la configuración de retorno.

Salta los efectos secundarios durante las vistas previas

Lakeflow Designer admite vistas previas mientras están en modo de diseño. Para los operadores que realizan llamadas a API externas o escriben en sistemas externos, añade una propiedad de configuración is_preview para poder omitir los efectos secundarios durante la vista previa. Los usuarios hacen clic en Ejecutar para ejecutar el operador con sus efectos secundarios. El operador de Python integrado no necesita esta propiedad, porque lee config["is_preview"] automáticamente. Consulta el modo de vista previa de Python.

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

Lakeflow Designer establece automáticamente este valor a true durante la vista previa. Compruébalo en la lógica para omitir los efectos secundarios:

# 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

Llama a APIs externas con conexiones HTTP de Unity Catalog

Para operadores definidos por el usuario basados en funciones SQL Unity Catalog que llaman APIs externas, utiliza conexiones HTTP Unity Catalog para almacenar credenciales 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'
);

A continuación, utilice la conexión en su UDF de SQL con la función http_request(). Para obtener más información, consulte Conexión a servicios HTTP externos.

Accede a los recursos del área de trabajo con WorkspaceClient

Para los operadores de python-run-function, pueden usar WorkspaceClient de Azure Databricks para acceder a los recursos del área de trabajo y a las API externas:

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

Recursos adicionales

  • Se actualizó por última vez el