How to Code a Data Pipeline Python

A Data Pipeline is an indispensable part of a data engineering workflow. It enables the extraction, transformation, and storage of data across disparate data sources and ensures that the right data is available at the right time. 

Python has emerged as a favorite tool for building such pipelines due to its scripting simplicity, extensive libraries, powerful frameworks, and strong community support.

This blog will explore how you can build a data pipeline python, including the options available for each step in the pipeline’s life cycle, tools, and frameworks, etc.

Why Data Pipeline Python?

Before Python, the data engineer had to work with different dialects and different scripting languages for each stage of a data pipeline. 

For example:

Data Pipeline Stages	Language/Dialects
Data Extraction Automation	Shell/Bash Scripting
Data Transformation	SQL/Java/Scala
Scheduling and Orchestrating	Cron Jobs (Linux/Windows)
Logging & Monitoring	Perl

Python gained popularity among data professionals due to its simplicity and readability of Python scripts. It allows rapid development and easy collaboration. Python ecosystem has evolved over time and has a tool, library, or framework for almost everything that a data professional needs.

Data professionals can now learn one scripting language and can work on every step of the lifecycle of a data pipeline, from extraction, ingestions, transformation, and validation to orchestration and monitoring. Python ecosystem has a rich set of libraries and frameworks that help us integrate with APIs and databases as either data sources or the destination of the pipeline. Its integration with popular cloud services such as AWS, Azure & GCP has further helped in creating cloud-native data pipelines.

Key Components of Data Pipeline:

Data traverses through several stages in the data engineering lifecycle, governed by several data pipeline components. Each component handles a specific part of the data processing workflow. 

Some of the key components of a data pipeline are briefly described below:

1. Data Ingestion
   This is one of the first stages of any data pipeline. Data sources are first integrated into the data pipeline and then the data is extracted and brought into the pipeline for further processing. Data sources can be either APIs, SQL/NoSQL Databases, flat files, etc. 
   Following are the available options while using Python for data ingestion for different data sources:

Data Sources	Python libraries
REST APIs	requests: A Python library to call REST APIs.
SQL Databases Postgres	psycopg2: Postgres adapter for Python.
SQL Databases MySQL	PyMySQL: Python MySQL Driver
NoSQL Databases	pymongo: Python driver for MongoDB
Flat Files	Pandas: A Python library to read flat files like CSV, JSON, etc.

2. Data transformation
   Once the raw data from data sources is ingested into the pipeline, it often needs cleaning and transformation to meet the desired format or schema. The pipeline’s data transformation component takes care of any data cleaning steps like data standardization, missing value imputation, aggregations, or any other required business-level complex calculations. 
   Python offers the following tools for data transformation steps:

Tool/Package	Description
Pandas	A Python package that offers powerful data structures for data analysis, time series, and statistical analysis.
Numpy	Fundamental package for complex mathematical computation.
PySpark	Python API for Spark Engine, a distributed computing framework.

3. Data validation
   The data validation component of the pipeline ensures that the data being delivered to the downstream storage unit or the data users meets the required quality standards.
   Python offers the following packages that data professionals widely use for validation and quality assertion:

Package	Description
Pydantic	A data type validation library for Python.
Cerberus	A lightweight data validation library for Python.
great-expectations	A data validation library for Python.

4. Data storage
   After data is transformed into a desired structure and format, it is stored in a database or a data lake. Python supports numerous storage solutions, from relational databases to data lake technology like AWS S3, Azure ADLS, etc.
   Common Python tools used for this component of the pipeline are:

Tool/Package	Description
SQLAlchemy	Python SQL Toolkit with ORM support.
Cloud SDKs	Python has separated SDKs for different cloud providers viz: AWS, Azure, and GCP.
boto3	A Python SDK to interact with AWS resources like S3 data lake.

5. Data orchestration and scheduling
   While working with complex data pipelines, multiple indigestion and transformation jobs may be dependent on other jobs. That is why job orchestration and scheduling are crucial for data pipeline automation. This helps ensure that job dependencies are met and that each job runs when it is supposed to run.
   Python supports the following tools for the orchestration and scheduling components of the pipeline:

Tool	Description
Apache Airflow	A widely used orchestration and scheduling tool that lets write jobs and dependencies using Python code.
Luigi	Python-based orchestration tool that is used mainly for building complex long-running batch jobs. It also supports task dependencies and failure recovery.
Perfect	A modern workflow orchestration tool for building modern data pipelines.

6. Data monitoring and logging
   There are times when ingestion or transformation jobs don’t run as planned, crash, or have performance bottlenecks. Thus, to track the performance metrics of the data pipeline, data monitoring, and logging components play a vital role in data engineering success.
   The Python logging module, along with monitoring tools like Prometheus, helps us monitor the data pipeline’s performance and tackle any bottlenecks.

Building a Data Pipeline in Python

The following is a step-by-step guide on how to build a data pipeline using Python.

Step 1: Installing Python Libraries

PIP is a Python package manager. We can use the package’s install command to install several of our project’s Python dependencies.

# using install command of pip package 
pip install -U pandas numpy sqlalchemy pydantic requests apache-airflow

Step 2: Ingesting data from a REST API

Python engineers use the requests library to call the REST API and fetch/ingest data into the pipeline.

# import requests library
import requests

# API URL to fetch data from
url = 'https://api.hevoexampleapi.com/data'

# Function that calls the API, fetches the data, and returns a JSON Object
def fetch_data_from_api(url):
    response = requests.get(url)
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"API request failed with status code {response.status_code}")

Step 3: Data Transformation

Pandas and numpy packages are used for data transformation.

# importing pandas and numpy used for data transformation
import pandas as pd
import numpy as np

# Function definition for data transformation logic
def transform_data(df):
    # Example transformation: Fill missing values
    df.fillna(0, inplace=True)
   
    # Example transformation: Add a calculated column
    df['adjusted_value'] = df['value'] * np.log(df['quantity'] + 1)
   
    return df

Step 4: Data Validation

Let us use Python’s Pydantic package for data validation.

# importing BaseModel from the pydantic package
from pydantic import BaseModel, ValidationError, conint

# Defining the data model based on the required data schema
class DataModel(BaseModel):
    name: str
    age: conint(gt=0)
    city: str

# Function definition for data validation across the data model.
def validate_data(data):
    try:
        return DataModel(**data)
    except ValidationError as e:
        print(e.json())
        return None

Step 5: Store data in the database

You can store the processed data in a database or cloud storage:

# importing create_engine function from sqlalchemy package
from sqlalchemy import create_engine

# function definition to write transformed data to the database
def store_data_in_db(df, connection_string, table_name):
    engine = create_engine(connection_string)
    df.to_sql(table_name, con=engine, if_exists='replace', index=False)

Step 6: Orchestrating the Pipeline with Apache Airflow

Apache Airflow is a powerful tool for orchestrating data pipelines. Here’s a simple example of an Airflow DAG (Directed Acyclic Graph) that orchestrates the pipeline:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetimeimport pandas as pd
import numpy as npfrom pydantic import BaseModel, ValidationError, conint
from sqlalchemy import create_engineimport requests

# API URL to fetch data from
url = 'https://api.hevoexampleapi.com/data'

# Function that calls the API, fetches the data, and returns a JSON Object
def fetch_data_from_api(url):
    response = requests.get(url)
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"API request failed with status code {response.status_code}")

# Function definition for data transformation logic
def transform_data(df):
    # Example transformation: Fill missing values
    df.fillna(0, inplace=True)
   
    # Example transformation: Add a calculated column
    df['adjusted_value'] = df['value'] * np.log(df['quantity'] + 1)
   
    return df

# Defining the data model based on the required data schema
class DataModel(BaseModel):
    name: str
    age: conint(gt=0)
    city: str

# Function definition for data validation across the data model.
def validate_data(data):
    try:
        return DataModel(**data)
    except ValidationError as e:
        print(e.json())
        return None

# function definition to write transformed data to the database
def store_data_in_db(df, connection_string, table_name):
    engine = create_engine(connection_string)
    df.to_sql(table_name, con=engine, if_exists='replace', index=False)

# Default arguments for the DAG
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 9, 7),
    'retries': 1,
}

# Defining a DAG
dag = DAG(
    'simple_data_pipeline',
    default_args=default_args,
    description='A simple data pipeline using Airflow',
    schedule_interval='@daily',
)

# Python operator for extraction task
extract_task = PythonOperator(
    task_id='extract',
    # Calling python function
    python_callable=fetch_data_from_api,
    dag=dag,
)

# Python operator for transform task
transform_task = PythonOperator(
    task_id='transform',
    # Calling Python function
    python_callable=transform_data,
    dag=dag,
)

# Python operator for data validation
validation_task = PythonOperatior(
    task_id ='data-validation',
    # Calling Python function
    python_callabe='validate_data_with_pydantic',
    dag=dag
)

# Python operator for load task
load_task = PythonOperator(
    task_id='load',
    # Calling python function
    python_callable=store_data_in_db,
    dag=dag,
)

# Airflow task dependency
extract_task >> transform_task >> validation_task >> load_task

This completes the whole Python data pipeline. It looks too technical, right? To write pipelines using Python, one must be a core Python programmer and understand the technicalities of data pipelines and how each component functions as one.

However, not all organizations have the resources or technical expertise to write a data pipeline from scratch. What if we don’t have to write a single line of code to create a data pipeline? Yes, you read it right. This is where a No-Code ETL tool like Hevo comes in. Let’s discuss this in brief:

Hevo: A No-Code ETL Tool:

Hevo Data is a fully managed, no-code ETL data pipeline platform. It helps us automate each process of a data pipeline, i.e., data ingestion, transformation, data validation, and storage, without having to write any code. It has been designed to help businesses of any size, with or without a technical team, build pipelines and democratize access to the power of data-based decisions.

Some of the features of Hevo are:

1. It has a user-friendly, drag-and-drop user interface that allows users to build and manage simple to complex data pipelines without writing any code.
2. It supports over 150+ integrations, from databases and SaaS applications to cloud storage and REST/SOAP APIs.
3. It offers real-time data ingestion without writing any complex real-time data ingestion logic.
4. It provides a huge library of pre-built transformation functions that are ready to use in our data pipeline. We can also customize it as per our needs.
5. It can automatically detect and map the schema of the data moving within the pipeline.
6. Powers automated data loading into all popular warehousing technologies like Amazon Redshift, BigQuery, Snowflake, S3, etc. All thanks to the rich data connectors Hevo supports.
7. The data pipeline scales automatically to meet the growing data needs of the organization.
8. It provides an interface to monitor the performance of the pipeline in real-time, along with an alerting feature.
9. The service and use cases of their data are well documented, making it easy to implement our use case. Their dedicated customer support makes troubleshooting even easier.

Learn More About:

How to Load Data into Python

Conclusion

Building a data pipeline has become a comparatively easy task after introducing Python in the data industry. Due to its scripting simplicity, extensive libraries, powerful frameworks, and strong community support, python has become data professionals’ favorite go-to tool for any data pipeline requirements. However, organizations nowadays look to simplify and accelerate their data ETL process without depending on a tech-savvy team and going through a tedious software development lifecycle for every data need. No-code ETL platforms like Hevo data become the best alternative for such a scenario. It provides the ease of use and integration capabilities essential for modern data workflows, enabling faster time-to-insight. 

Frequently Asked Questions on Data Pipeline Python