Write logic for Lakeflow Designer user-defined operators

The operator logic is the code that runs when a user-defined operator executes. For a python-run-function operator, it's an inline Python run() function. For a uc-udf or uc-udtf operator, it's a Unity Catalog function written in SQL or Python.

Writing the logic is one step in building a shared user-defined operator. After you write and test the logic, you describe the operator in YAML so Lakeflow Designer knows how to show it in the UI, then register it so it appears in the Operators pane. To see all of the steps together for one operator, follow a tutorial such as Tutorial: Compound interest calculator operator.

Write a python-run-function operator

Every python-run-function operator must define a run() function:

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config: User-configured values from the UI, keyed by property name.
  • inputs: Input DataFrames, keyed by input port name.
  • spark: The active SparkSession.
  • Returns: A dictionary mapping output port name values to DataFrames.

The following example filters rows from an input DataFrame:

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

To test the function before you register it, call it in a notebook with sample data:

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

If your operator requires external pip packages, add the environment field to the YAML:

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

Write a UDF or UDTF operator

You can write Unity Catalog functions in SQL or Python. Python functions are wrapped in a SQL CREATE FUNCTION statement:

SQL function:

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

Python function (wrapped in 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 process a single value at a time and return a calculated value. UDTFs process tables row by row and can maintain state across all rows. Use uc-udf for column-level transforms and uc-udtf for operations like ML clustering or aggregation.

A UDTF handler class defines eval(), which runs for each input row. Optionally, it also defines __init__() to set up state and terminate() to yield final results after all rows. See Python user-defined table functions (UDTFs) in 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

UDTF return tables must have fixed, explicit types. You can't reference input column types in the return configuration.

Skip side effects during previews

Lakeflow Designer supports previews while in design mode. For operators that call external APIs or write to external systems, add an is_preview config property so you can skip side effects during preview. Users click Run to run the operator with its side effects. The built-in Python operator doesn't need this property, because it reads config["is_preview"] automatically. See Python preview mode.

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

Lakeflow Designer automatically sets this value to true during preview. Check it in your logic to skip side effects:

# 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

Call external APIs with Unity Catalog HTTP connections

For user-defined operators based on SQL Unity Catalog functions that call external APIs, use Unity Catalog HTTP connections to securely store credentials:

CREATE CONNECTION my_api_connection TYPE HTTP OPTIONS (
  host 'https://api.example.com',
  port '443',
  base_path '/v1/',
  bearer_token 'your-token-here'
);

Then use the connection in your SQL UDF with the http_request() function. For details, see Connect to external HTTP services.

Access workspace resources with WorkspaceClient

For python-run-function operators, you can use the Azure Databricks WorkspaceClient to access workspace resources and external APIs:

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

Additional resources

  • Last updated on