Write Your First Pipeline Node

first_nodes.py

"""
first_nodes.py

used in video https://youtu.be/-gEwU-MrPuA
Full Post 👉 https://waylonwalker.com/kedro-your-first-nodes/
"""

from kedro.pipeline import node


def create_range():
    return range(100)


make_range = node(func=create_range, inputs=None, outputs="range")


def square(range):
    return [i ** 2 for i in range]


square_range = node(func=square, inputs="range", outputs="range_squared")


def concat(range, range_two):
    return [*range, *range_two]


# concat_ranges = node(func=concat, inputs=["range", "range_squared"], outputs="concat")

concat_ranges = node(
    func=concat,
    inputs={"range": "range", "range_two": "range_squared"},
    outputs="concat",
)

if __name__ == "__main__":
    from kedro.pipeline import Pipeline
    from kedro.runner import SequentialRunner
    from kedro.io import DataCatalog

    pipeline = Pipeline([make_range, square_range, concat_ranges])
    runner = SequentialRunner()
    result = runner.run(pipeline, DataCatalog())
    print(result)