Create a custom pipeline

Adding a new pipeline in pamflow is simple by using the Kedro CLI tools. Here is the straightforward, step-by-step process:

1. Generate the Pipeline

Open your terminal in the root of the pamflow repository and run the following command:

kedro pipeline create [pipeline_name]

(Replace [pipeline_name] with your desired name, e.g., preprocessing.)

What this does:

• Creates a folder in src/pamflow/pipelines/[pipeline_name].

• Generates template files: nodes.py (for your logic) and pipeline.py (for the structure).

• Creates a parameters file in conf/base/parameters_[pipeline_name].yml.

• Creates a test template in src/tests/pipelines/[pipeline_name]/.

---

2. Define your Logic in nodes.py

Open src/pamflow/pipelines/[pipeline_name]/nodes.py and write your Python functions.

def my_first_node(data):
    # Your transformation logic here
    return data.upper()

---

3. Connect the Nodes in pipeline.py

Open src/pamflow/pipelines/[pipeline_name]/pipeline.py. Use the node() function to map your Python functions to inputs and outputs defined in your Data Catalog.

from kedro.pipeline import Pipeline, node
from .nodes import my_first_node

def create_pipeline(**kwargs):
    return Pipeline([
        node(
            func=my_first_node,
            inputs="raw_data_from_catalog",
            outputs="processed_data_output",
            name="transform_node",
        ),
    ])

---

4. Register the Pipeline

Open src/pamflow/pipeline_registry.py and update it in three places:

1. Add the import at the top of the file:

from pamflow.pipelines.[pipeline_name] import pipeline as my_pipeline

2. Instantiate the pipeline inside register_pipelines():

def register_pipelines() -> Dict[str, Pipeline]:
    my_pipeline_instance = my_pipeline.create_pipeline()
    ...

3. Add the pipeline to the return dictionary:

    return {
        "__default__": pamflow_pipeline,
        ...
        "[pipeline_name]": my_pipeline_instance,
    }

Note

Keep "__default__" pointing to pamflow_pipeline. If you replace it with your pipeline, pamflow run will no longer run the standard workflow. To run your pipeline together with the standard one, add it to the default instead: "__default__": pamflow_pipeline + my_pipeline_instance.

---

5. Verify and Run

To make sure Kedro sees your new pipeline, list them first:

kedro registry list

Then, run only your new pipeline to test it:

pamflow run --pipeline [pipeline_name]

---

Troubleshooting Tip: Check your Nodes

If the run fails, you can use the following command to see exactly how Kedro is interpreting your new pipeline’s structure:

kedro pipeline describe [pipeline_name]