Python Tasks¶

The Context Model lets you run Python Tasks: Python code that reads input tables, transforms or generates data, and writes the result back to one or more tables. pycelonis.ml_workbench.transformation provides a small, Foundry-style framework to author these tasks directly in a Jupyter Notebook in the ML Workbench, so you can iterate interactively before scheduling the notebook as a Python Task.

The framework takes care of:

• Creating and tearing down a SparkSession connected to the Storium Spark Proxy
• Wiring your function's parameters to the configured input/output tables by name
• Providing a dry_run mode so you can safely test your logic without writing any data

Prerequisites¶

• pyspark needs to be installed, e.g. via pip install pycelonis[mlwb].
• You need a Context Model that contains the tables you want to read, or a Data Pool containing sample input tables.
• Interactive testing is intended to run in ML Workbench, where the Spark connection is automatically set up using your credentials.

Tutorial¶

Writing a @transformation for a Python Task boils down to three steps:

1. Configure TransformationConfig with the schema and tables for interactive notebook testing.
2. Decorate a function with @transformation and let it receive its input/output tables as parameters.
3. Call the function.

1. Configure the transformation¶

TransformationConfig.set_defaults(...) sets the schema, input and output tables, and dry-run behavior for interactive testing. schema and output_tables are required; input_tables may be empty for a task that generates data without reading a table. dry_run defaults to True and previews output DataFrames instead of writing them.

Once this notebook is published and scheduled as a Python Task, the Context Model injects the real task configuration (schema, tables, and dry_run) automatically. set_defaults never overrides values provided that way, so this cell is safe to leave in the notebook: it becomes a no-op in a triggered task and can be re-run while iterating interactively.

In [ ]:

from pycelonis.ml_workbench.transformation import TransformationConfig

# Configuration used for interactive testing. In a triggered Python Task, these values
# are supplied through execution parameters and this call becomes a no-op.
TransformationConfig.set_defaults(
    schema="<schema_id>",  # Context Model branch schema ID or Data Pool ID
    input_tables=["input_table", "other_schema.imported_table"],
    output_tables=["output_table"],
    dry_run=True,  # The platform sets this to False for a triggered Python Task
)

from pycelonis.ml_workbench.transformation import TransformationConfig # Configuration used for interactive testing. In a triggered Python Task, these values # are supplied through execution parameters and this call becomes a no-op. TransformationConfig.set_defaults( schema="<schema_id>", # Context Model branch schema ID or Data Pool ID input_tables=["input_table", "other_schema.imported_table"], output_tables=["output_table"], dry_run=True, # The platform sets this to False for a triggered Python Task )

2. Define the transformation¶

@transformation inspects your function's parameters and matches each one to a configured input or output table:

• A parameter is matched automatically if its name equals the table name, or - for a table imported from another schema (other_schema.imported_table) - the part after the last dot (imported_table).
• If the parameter name differs, or the table name is not a valid Python identifier, declare it explicitly: @transformation(<param_name>="[<schema>.]<table_name>").
• If multiple configured tables share an unqualified name, use explicit mappings to disambiguate them.
• Path segments containing characters other than ASCII letters, digits, or underscores are escaped automatically for Spark. Do not add backticks to configured table paths.

Input parameters are wrapped in an Input object (.dataframe(), .count(), .schema(), .columns(), .show()), and output parameters in an Output object (.write_dataframe(df)).

In [ ]:

from pycelonis.ml_workbench.transformation import transformation


@transformation()
def process(input_table, imported_table, output_table):
    df = input_table.dataframe()
    print(f"Read {input_table.count()} rows from '{input_table.table_name}'")
    print(f"Read {imported_table.count()} rows from '{imported_table.table_name}'")

    # ... your transformation logic goes here ...

    output_table.write_dataframe(df)

from pycelonis.ml_workbench.transformation import transformation @transformation() def process(input_table, imported_table, output_table): df = input_table.dataframe() print(f"Read {input_table.count()} rows from '{input_table.table_name}'") print(f"Read {imported_table.count()} rows from '{imported_table.table_name}'") # ... your transformation logic goes here ... output_table.write_dataframe(df)

3. Run the transformation¶

Calling the decorated function creates the SparkSession, wires up the tables, runs your code, and always stops the session afterwards - even if your code raises an exception.

Because dry_run=True above, output_table.write_dataframe(df) calls df.show() instead of writing the table, so you can safely test your logic first. With dry_run=False, it overwrites the target table's data and schema.

In [ ]:

process()

process()

From notebook to Python Task¶

Once you are happy with the result:

1. Publish this ML Workbench.
2. Create a Transformation resource in the ML Workbench UI.
3. Select that ML Workbench Transformation when creating a Python Task in the Context Model.
4. Ensure the input and output table names match between the Python Task and TransformationConfig or the explicit @transformation(...) mappings.

The Context Model then provides TransformationConfig (schema, tables, and dry_run=False) through the task's execution parameters, so the interactive defaults above are ignored.

Conclusion¶

You have learned how to configure TransformationConfig, wire up tables with @transformation, and safely test a Python Task interactively using dry_run before scheduling it in the Context Model.