In this article, we cover how to use pipeline patterns in python data engineering projects. Here are the steps: Functional pipeline fastcore Install fastcore Creating pipeline using fastcore Dynamic pipeline using fastcore Let's get into it! Functional pipeline The functional pipeline is a design pattern mostly used in the functional programming paradigm, where data flows through a sequence of stages and the output of the previous stage is the input of the next. Each step can be thought of as a filter operation that transforms the data in some way. This pattern is most suitable for map, filter and reduces operations. It also provides a clean, readable and more sustainable code in data engineering projects. For example, let's take an input text which has to go through a series of transformations, Remove white spaces Remove special characters Lowercase all letters and finally produces output. These pipeline functions are simplified to demonstrate the use case, In a real-life scenario, it would be a lot more complex. Let's create the simple transformation functions. re output = string.replace( , )
    print( ) output output = re.sub( , , string)
    print( ) output output = string.lower()
    print( ) output #pipeline_functions.py import : def remove_spaces (string) ' ' '' f""" () ==> """ {remove_spaces.__name__} {output} return : def remove_special_chars (string) "[^A-Za-z0-9]" "" f""" () ==> """ {remove_special_chars.__name__} {output} return : def lowercase (string) f""" () ==> """ {lowercase.__name__} {output} return fastcore fastcore is a utility that has a lot of python goodies to make coding faster, easier, and more maintainable. It borrows some ideas from other languages like Julia, Ruby, and Haskell. It also adds functional programming patterns, simplified parallel processing, and a lot more. For our pipeline implementation, we will be using fastcore transform module. Do checkout  -> https://fastcore.fast.ai/ Install fastcore $ pip install fastcore Creating pipeline using fastcore fastcore.transform Pipeline pipeline_functions remove_spaces, remove_special_chars, lowercase pipe = Pipeline([remove_spaces, remove_special_chars, lowercase]) output = pipe(input_string)

    print( ) __name__ == :
    text = input( )
    main(text) #main.py from import from import : def main (input_string) # Creates a pipeline with a list of functions # Invokes pipeline f"""output ==> """ {output} if '__main__' "Enter input string: " Run the program $ python main.py
Enter input string: Hello World!
remove_spaces() ==> HelloWorld!
remove_special_chars() ==> HelloWorld
lowercase() ==> helloworld
output ==> helloworld As you can see that entered input text gets passed through the pipeline from left to right order and manipulates input text in each step and returns the final output. We can even go one step further to have more dynamic pipeline functions by getting pipeline functions at runtime. So this list of functions can also be serialized and persisted for later use. Dynamic pipeline sys fastcore.transform Pipeline pipeline_functions remove_spaces, remove_special_chars, lowercase pipe = Pipeline([globals()[func] func pipe_funcs]) output = pipe(input_string)

    print( ) __name__ == :
    text = input( )
    funcs = sys.argv[ :]
    main(text, funcs) #main_dynamic.py import from import from import : def main (input_string, pipe_funcs) # Creates a pipeline with a list of functions using using globals() for in # Invokes pipeline f"""output ==> """ {output} if '__main__' "Enter input string: " 1 Run the program $ python main_dynamic.py remove_spaces lowercase 
Enter input string: Hello World 123$    
remove_spaces() ==> HelloWorld123$
lowercase() ==> helloworld123$
output ==> helloworld123$ Conclusion So the pipeline pattern implementation in data engineering components makes it easier to write complex data processing operations. And fastcore utility makes it even better. If you have found this tutorial helpful, or have any suggestions do let me know! Previously published at https://sureshdsk.dev/pipeline-pattern-in-python-data-engineering

Introducing Tour My Pull Request For Github Extension

https://bio.link/sureshdsk

Read My Stories

Too Long; Didn't Read

Make resilience your competitive advantage

How To Create a Python Data Engineering Project with a Pipeline Pattern

About Author

Comments

TOPICS

THIS ARTICLE WAS FEATURED IN

Related Stories

Untitled Story

Introducing Tour My Pull Request For Github Extension

106 Stories To Learn About Clean Code

100 Pieces of Programming Advice from the Book Clean Code by Robert Martin

110 Stories To Learn About Refactoring

3 Things Coding And Prose Writing Have In Common According To Cory House

Introducing Tour My Pull Request For Github Extension

106 Stories To Learn About Clean Code

100 Pieces of Programming Advice from the Book Clean Code by Robert Martin

110 Stories To Learn About Refactoring

3 Things Coding And Prose Writing Have In Common According To Cory House

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps