Write a Python transform as collection code

Use a Python or SQL Code asset when a built-in transform cannot express the logic clearly. A code node references a reusable Code asset in the selected collection; it is not a pipeline-local script.

In this lesson, you keep the lesson 3 result unchanged and build a separate pipeline that filters high-value orders.

What the code does

The code keeps orders whose amount is at least 10,000 and renames the source date column for the output schema.

For this single condition, the built-in Select/Filter transform would be simpler. The exercise uses Python so you can learn the code node's input and output contract.

Create the output dataset

1. Open the collection that contains src_postgres_orders.
2. Select Add item, point to Dataset, and select Table.
3. Enter mart_orders_filtered for Name and select Next.
4. Configure these attributes under Schema.

5. Select Create and confirm that mart_orders_filtered appears in the collection.

Use an existing dataset only when its schema is identical. Changing the schema of a dataset that already contains data can make later writes fail validation.

Open a new pipeline

1. Select Data → Pipelines in the sidebar.
2. Select Create in the upper-right.
3. In Select Pipeline Collection, choose the collection containing the input and output datasets.

Create a new pipeline instead of modifying lesson 3 so both results remain available.

Create the Python Code asset

1. Drag Quick Add → Code → Python onto the canvas.
2. Name it filter_high_value_orders.
3. Enter this code.

import polars as pl

def run(src_postgres_orders, *, options=None, contexts=None):
    output = (
        src_postgres_orders
        .with_columns(
            pl.col("amount").cast(pl.Int64),
            pl.col("ord_dt_str").alias("order_dt_str"),
        )
        .filter(pl.col("amount") >= 10000)
        .select(["order_id", "order_dt_str", "amount"])
    )
    return {"mart_orders_filtered": output}

The runtime passes connected inputs as keyword arguments. Therefore, the function argument src_postgres_orders must exactly match the input port name. The returned key mart_orders_filtered must exactly match the output port name. A mismatch causes an unexpected keyword argument or output-key error.

Connect the pipeline

1. Drag src_postgres_orders from the collection section onto the canvas.
2. Connect it to the left handle of filter_high_value_orders.
3. Drag mart_orders_filtered onto the canvas.
4. Connect the code node's right handle to mart_orders_filtered.
5. In the code node's Options tab, confirm that the input is src_postgres_orders, exactly matches the first run argument, and uses incremental read mode.
6. Confirm the mart_orders_filtered output and append write mode.

src_postgres_orders → filter_high_value_orders → mart_orders_filtered

Save, run, and verify

1. Select Save and enter mart_orders_filtered_pipeline.
2. Select Run now.
3. Confirm a successful run in the run history.
4. Open mart_orders_filtered and select Data.

The result should contain these two rows. The 8,900 order is excluded.

Self-check

• Is filter_high_value_orders saved as a Code asset in the selected collection?
• Are the nodes connected dataset → code → dataset?
• Do the function argument and returned key match the pipeline ports?
• Does the successful output contain only the two orders of at least 10,000?

Next lesson

Next, you configure a schedule for the batch pipeline.