Transformstransforms-pythontransforms.apitransform_pandas

transforms.api.transform_pandas

transforms.api.transform_pandas(output, **inputs)

To use the pandas library, you must add pandas as a run dependency in your meta.yml file. For more information, refer to the documentation ↗.

If you prefer using Polars DataFrames, use the transform_polars decorator to register your compute function as a Polars transform.

The transform_pandas decorator is used to construct a Transform object from a compute function that accepts and returns pandas.DataFrame ↗ objects. This decorator is similar to the transform_df() decorator, however the pyspark.sql.DataFrame ↗ objects are converted to pandas.DataFrame ↗ object before the computation, and converted back afterwards.

Copied!1
2
3
4
5
6
7
8
>>> @transform_pandas(
...     Output('/path/to/output/dataset'),  # An unnamed Output spec
...     first_input=Input('/path/to/first/input/dataset'),
...     second_input=Input('/path/to/second/input/dataset'),
... )
... def my_compute_function(first_input, second_input):
...     # type: (pandas.DataFrame, pandas.DataFrame) -> pandas.DataFrame
...     return first_input.concat(second_input)

You can wrap a pandas transform with the incremental decorator to run it incrementally. When @incremental wraps transform_pandas, inputs are read using the added mode and the write mode is determined by @incremental. If you need to read or write using non-default modes, use the transform() decorator instead.

Copied!1
2
3
4
5
6
7
8
9
>>> @incremental()
... @transform_pandas(
...     Output('/path/to/output/dataset'),
...     first_input=Input('/path/to/first/input/dataset'),
... )
... def compute(first_input):
...     # type: (pandas.DataFrame) -> pandas.DataFrame
...     # "added" rows are provided; output write behavior is controlled by @incremental
...     return first_input

Note that transform_pandas should only be used on datasets that can fit into memory. If you have larger datasets that you wish to filter down before converting to pandas you should write your transformation using the transform_df() decorator and the pyspark.sql.SparkSession.createDataFrame() ↗ method.

Copied!1
2
3
4
5
6
7
8
9
10
>>> @transform_df(
...     Output('/path/to/output/dataset'),  # An unnamed Output spec
...     first_input=Input('/path/to/first/input/dataset'),
...     second_input=Input('/path/to/second/input/dataset'),
... )
... def my_compute_function(ctx, first_input, second_input):
...     # type: (pyspark.sql.DataFrame, pyspark.sql.DataFrame) -> pyspark.sql.DataFrame
...     pd = first_input.filter(first_input.county == 'UK').toPandas()
...     # Perform pandas operations on a subset of the data before converting back to a PySpark DataFrame
...     return ctx.spark_session.createDataFrame(pd)

Parameters:
- output (Output) – The single Output spec for the transform.
- **inputs (Input) – kwargs comprised of named Input specs.