transforms.api.transform

class transforms.api.transform(**kwargs)

Wrap a compute function as a Transform object.

The transform decorator is used to construct a Transform object from a compute function. The names used for inputs, outputs or other parameters should be the function arguments of the wrapped compute function. At compute-time, parameters are instantiated to specific objects defined by each parameter type.

Copied!
1
2
3
4
5
6
7
8
9
10
>>> @transform(
...     first_input=Input('/path/to/first/input/dataset'),
...     second_input=Input('/path/to/second/input/dataset'),
...     first_output=Output('/path/to/first/output/dataset'),
...     second_output=Output('/path/to/second/output/dataset'),
... )
... def my_compute_function(first_input, second_input, first_output, second_output):
...     # type: (TransformInput, TransformInput, TransformOutput, TransformOutput) -> None
...     first_output.write_dataframe(first_input.dataframe())
...     second_output.write_dataframe(second_input.dataframe())

• Parameters: **kwargs (Param) – kwargs comprised of named Param or subclasses.

Optional context parameter

You can optionally accept a context object in your compute function by adding a ctx argument. For regular Spark transforms, the context is a transforms.api.TransformContext, which exposes runtime information and utilities such as abort_job(), auth_header, environment, fallback_branches, parameters, and spark_session.

Copied!
1
2
3
4
5
6
7
8
9
10
11
>>> @transform(
...     my_input=Input('/path/to/input'),
...     my_output=Output('/path/to/output'),
... )
... def my_compute_function(ctx, my_input, my_output):
...     # ctx is a TransformContext
...     df = my_input.dataframe()
...     # Optionally use ctx, for example, to access parameters or abort the job
...     if ctx.parameters.get('abort'):
...         ctx.abort_job()
...     my_output.write_dataframe(df)

For lightweight transforms created with transform.using(...), the injected context is a transforms.api.LightweightContext, which exposes methods and properties such as abort_job(), auth_header, and is_incremental.

Copied!
1
2
3
4
5
6
7
8
9
>>> @transform.using(
...     my_input=Input('/input'),
...     my_output=Output('/output')
... )
... def compute_func(ctx, my_input, my_output):
...     if not ctx.is_incremental:
...         # End execution and abort all output transactions
...         ctx.abort_job()
...     my_output.write_pandas(my_input.pandas())

Incremental execution

To run a transform incrementally, wrap it with the incremental decorator. The incremental decorator uses output build history to determine the previous state and converts TransformInput, TransformOutput, and TransformContext to their incremental counterparts.

Copied!
1
2
3
4
5
6
7
8
# Spark transform, executed incrementally
@incremental()
@transform(
    first_input=Input('/path/to/first/input/dataset'),
    first_output=Output('/path/to/first/output/dataset'),
)
def my_compute_function(first_input, first_output):
    first_output.write_dataframe(first_input.dataframe())

Copied!
1
2
3
4
5
6
7
8
# Lightweight transform, executed incrementally
@incremental()
@transform.using(
    my_input=Input('/input'),
    my_output=Output('/output')
)
def compute_func(my_input, my_output):
    my_output.write_pandas(my_input.pandas())

classmethod using(**kwargs)

Construct a lightweight transform via the transform class.

A lightweight transform is a transform that runs without Spark, on a single node. Lightweight transforms are faster and more cost-effective for small to medium-sized datasets. Lightweight transforms also provide more methods for accessing datasets; however, they only support a subset of the API of a regular transform, including pandas, Polars (via transform_polars), and the filesystem API. Spark profiles are not supported for lightweight transforms.

This method accepts only input/output parameters. Resource requirements and container settings can be configured by chaining additional method calls.

For more information, see the Transforms Python overview. For Polars DataFrame-based transformations, see transform_polars.

• Parameters: **kwargs – Named LightweightInput or LightweightOutput objects for the transform’s inputs and outputs.
• Returns: A builder object that can be configured with additional method calls.

Examples

Copied!
1
2
3
4
5
6
7
>>> # Basic usage with default resources
>>> @transform.using(
...     my_input=Input('/input'),
...     my_output=Output('/output')
... )
... def compute_func(my_input, my_output):
...     my_output.write_pandas(my_input.pandas())

Copied!
1
2
3
4
5
6
7
8
9
10
>>> # Configure resources
>>> @transform.using(
...     my_input=Input('/input'),
...     my_output=Output('/output')
... ).with_resources(
...     cpu_cores=4,
...     memory_gb=8
... )
... def compute_func(my_input, my_output):
...     my_output.write_pandas(my_input.pandas())

Copied!
1
2
3
4
5
6
7
8
9
10
>>> # Configure container
>>> @transform.using(
...     my_output=Output('/output')
... ).with_container(
...     container_image='my-image',
...     container_tag='0.0.1'
... )
... def run_data_generator_executable(my_output):
...     os.system('$USER_WORKING_DIR/data_generator')
...     my_output.write_table(pd.read_csv('data.csv'))