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

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