Data Engineering Best Practices: Building Scalable and Reliable Data Pipelines

Andrew • Jul 1, 2025 • Data Pipelines , ETL , ELT , Data Quality , Data Governance , Apache Airflow , dbt , Streaming Data , Data Lakes , Data Warehouses

12 min read 2468 words

Data engineering has become a critical discipline as organizations increasingly rely on data to drive decision-making and power advanced analytics. Building robust data pipelines that reliably collect, transform, and deliver data requires careful attention to architecture, processes, and tooling. However, many organizations struggle with data quality issues, pipeline reliability problems, and scalability challenges that prevent them from fully leveraging their data assets.

This comprehensive guide explores data engineering best practices, covering pipeline architecture, ETL/ELT processes, data quality, governance, and modern tools. Whether you’re building new data infrastructure or improving existing pipelines, these insights will help you create scalable, reliable, and maintainable data systems that deliver high-quality data to your organization’s analytical and operational workloads.

Data Pipeline Architecture

Architectural Patterns

Foundational approaches to data pipeline design:

Batch Processing:

Processing data in scheduled intervals
Handling large volumes efficiently
Optimizing for throughput over latency
Implementing idempotent operations
Managing dependencies between jobs

Stream Processing:

Processing data in near real-time
Handling continuous data flows
Implementing windowing strategies
Managing state and checkpointing
Ensuring exactly-once processing

Lambda Architecture:

Combining batch and streaming layers
Providing both accurate and real-time views
Managing duplicate processing logic
Reconciling batch and speed layers
Optimizing for different access patterns

Kappa Architecture:

Unifying batch and streaming with a single path
Simplifying maintenance with one codebase
Leveraging stream processing for all workloads
Reprocessing historical data through streams
Reducing architectural complexity

Data Mesh:

Decentralizing data ownership
Treating data as a product
Implementing domain-oriented architecture
Providing self-serve data infrastructure
Establishing federated governance

Example Lambda Architecture:

┌───────────────┐
│               │
│  Data Sources │
│               │
└───────┬───────┘
        │
        ▼
┌───────────────┐     ┌───────────────┐
│               │     │               │
│  Batch Layer  │     │ Speed Layer   │
│               │     │               │
└───────┬───────┘     └───────┬───────┘
        │                     │
        ▼                     ▼
┌───────────────┐     ┌───────────────┐
│               │     │               │
│  Batch Views  │     │ Real-time     │
│               │     │ Views         │
└───────┬───────┘     └───────┬───────┘
        │                     │
        └─────────┬───────────┘
                  │
                  ▼
          ┌───────────────┐
          │               │
          │  Serving      │
          │  Layer        │
          │               │
          └───────────────┘

ETL vs. ELT

Comparing transformation approaches:

ETL (Extract, Transform, Load):

Transformation before loading to target
Data cleansing outside the data warehouse
Typically uses specialized ETL tools
Better for complex transformations with limited compute
Reduced storage requirements in target systems

ELT (Extract, Load, Transform):

Loading raw data before transformation
Leveraging data warehouse compute power
Enabling exploration of raw data
Simplifying pipeline architecture
Supporting iterative transformation development

Hybrid Approaches:

Light transformation during extraction
Heavy transformation in the warehouse
Preprocessing for specific use cases
Optimizing for different data types
Balancing performance and flexibility

When to Choose ETL:

Limited data warehouse resources
Complex transformations requiring specialized tools
Strict data privacy requirements
Legacy system integration
Real-time transformation needs

When to Choose ELT:

Modern cloud data warehouses with scalable compute
Exploratory analytics requirements
Evolving transformation requirements
Large volumes of structured or semi-structured data
Self-service analytics environments

Orchestration

Managing pipeline workflows and dependencies:

Orchestration Requirements:

Dependency management
Scheduling and triggering
Error handling and retries
Monitoring and alerting
Resource management

Apache Airflow:

DAG-based workflow definition
Python-based configuration
Rich operator ecosystem
Extensive monitoring capabilities
Strong community support

Example Airflow DAG:

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.transfers.s3_to_redshift import S3ToRedshiftOperator
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.providers.postgres.operators.postgres import PostgresOperator

default_args = {
    'owner': 'data_engineering',
    'depends_on_past': False,
    'start_date': datetime(2025, 7, 1),
    'email': ['[email protected]'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'sales_data_pipeline',
    default_args=default_args,
    description='Load and transform sales data',
    schedule_interval='0 2 * * *',
    catchup=False,
    tags=['sales', 'production'],
)

# Check if data is available
check_data_available = S3KeySensor(
    task_id='check_data_available',
    bucket_key='sales/{{ ds }}/*.csv',
    wildcard_match=True,
    bucket_name='sales-data',
    aws_conn_id='aws_default',
    timeout=60 * 30,
    poke_interval=60,
    dag=dag,
)

# Load data from S3 to Redshift
load_to_redshift = S3ToRedshiftOperator(
    task_id='load_to_redshift',
    schema='raw',
    table='sales',
    s3_bucket='sales-data',
    s3_key='sales/{{ ds }}/',
    redshift_conn_id='redshift_default',
    aws_conn_id='aws_default',
    copy_options=[
        "DELIMITER ','",
        "IGNOREHEADER 1",
        "DATEFORMAT 'auto'",
    ],
    method='REPLACE',
    dag=dag,
)

# Transform data
transform_data = PostgresOperator(
    task_id='transform_data',
    postgres_conn_id='redshift_default',
    sql="""
    INSERT INTO analytics.daily_sales_summary
    SELECT 
        date_trunc('day', sale_timestamp) as sale_date,
        product_id,
        SUM(quantity) as total_quantity,
        SUM(amount) as total_amount,
        COUNT(DISTINCT customer_id) as unique_customers
    FROM raw.sales
    WHERE DATE(sale_timestamp) = '{{ ds }}'
    GROUP BY 1, 2
    """,
    dag=dag,
)

# Define task dependencies
check_data_available >> load_to_redshift >> transform_data

Other Orchestration Tools:

Prefect
Dagster
Argo Workflows
Luigi
AWS Step Functions

Orchestration Best Practices:

Define clear task boundaries
Implement proper error handling
Use parameterization for reusability
Monitor pipeline performance
Implement CI/CD for pipeline code

Data Quality and Testing

Data Quality Dimensions

Key aspects of data quality to monitor:

Completeness:

Checking for missing values
Validating required fields
Monitoring record counts
Comparing against expected totals
Tracking data arrival

Accuracy:

Validating against known values
Cross-checking with reference data
Implementing business rule validation
Detecting anomalies and outliers
Verifying calculations

Consistency:

Checking for contradictory values
Validating referential integrity
Ensuring uniform formats
Comparing across systems
Monitoring derived values

Timeliness:

Tracking data freshness
Monitoring pipeline latency
Validating timestamp sequences
Alerting on delayed data
Measuring processing time

Uniqueness:

Detecting duplicates
Validating primary keys
Checking composite uniqueness constraints
Monitoring merge operations
Tracking deduplication metrics

Testing Strategies

Approaches to validate data quality:

Unit Testing:

Testing individual transformation functions
Validating business logic
Checking edge cases
Mocking dependencies
Automating with CI/CD

Example Python Unit Test:

import unittest
from transformations import calculate_revenue

class TestTransformations(unittest.TestCase):
    
    def test_calculate_revenue(self):
        # Test normal case
        input_data = {
            'quantity': 5,
            'unit_price': 10.0,
            'discount_percentage': 20.0
        }
        expected = 40.0  # 5 * 10 * (1 - 0.2)
        self.assertEqual(calculate_revenue(input_data), expected)
        
        # Test zero quantity
        input_data = {
            'quantity': 0,
            'unit_price': 10.0,
            'discount_percentage': 20.0
        }
        expected = 0.0
        self.assertEqual(calculate_revenue(input_data), expected)
        
        # Test no discount
        input_data = {
            'quantity': 5,
            'unit_price': 10.0,
            'discount_percentage': 0.0
        }
        expected = 50.0
        self.assertEqual(calculate_revenue(input_data), expected)

Integration Testing:

Testing complete data flows
Validating end-to-end processes
Using test environments
Simulating production scenarios
Checking system interactions

Data Quality Rules:

Implementing schema validation
Defining value constraints
Setting threshold-based rules
Creating relationship rules
Establishing format validation

Example dbt Tests:

# Schema.yml for dbt tests
version: 2

models:
  - name: orders
    description: "Cleaned orders table"
    columns:
      - name: order_id
        description: "Primary key of the orders table"
        tests:
          - unique
          - not_null
      
      - name: customer_id
        description: "Foreign key to customers table"
        tests:
          - not_null
          - relationships:
              to: ref('customers')
              field: customer_id
      
      - name: order_date
        description: "Date when the order was placed"
        tests:
          - not_null
          - dbt_utils.date_in_range:
              min_date: '2020-01-01'
              max_date: '{{ current_date() }}'
      
      - name: status
        description: "Current status of the order"
        tests:
          - accepted_values:
              values: ['pending', 'shipped', 'delivered', 'returned', 'cancelled']

Monitoring and Alerting:

Setting up data quality dashboards
Implementing anomaly detection
Creating alerting thresholds
Tracking quality metrics over time
Establishing incident response procedures

Data Observability

Gaining visibility into data systems:

Observability Pillars:

Freshness monitoring
Volume tracking
Schema changes
Lineage visualization
Distribution analysis

Example Freshness Monitoring Query:

-- PostgreSQL query to monitor data freshness
WITH source_freshness AS (
    SELECT
        'sales_data' AS source_name,
        MAX(created_at) AS last_record_time,
        NOW() - MAX(created_at) AS staleness,
        CASE
            WHEN NOW() - MAX(created_at) > INTERVAL '1 day' THEN 'critical'
            WHEN NOW() - MAX(created_at) > INTERVAL '6 hours' THEN 'warning'
            ELSE 'healthy'
        END AS status
    FROM raw_data.sales
    
    UNION ALL
    
    SELECT
        'customer_data' AS source_name,
        MAX(created_at) AS last_record_time,
        NOW() - MAX(created_at) AS staleness,
        CASE
            WHEN NOW() - MAX(created_at) > INTERVAL '7 days' THEN 'critical'
            WHEN NOW() - MAX(created_at) > INTERVAL '3 days' THEN 'warning'
            ELSE 'healthy'
        END AS status
    FROM raw_data.customers
)
SELECT
    source_name,
    last_record_time,
    staleness,
    status
FROM source_freshness
ORDER BY 
    CASE status
        WHEN 'critical' THEN 1
        WHEN 'warning' THEN 2
        ELSE 3
    END,
    staleness DESC;

Observability Tools:

Great Expectations
Monte Carlo
Datadog
Prometheus with custom exporters
dbt metrics

Data Transformation

Modern ELT with dbt

Implementing analytics engineering best practices:

dbt Core Concepts:

Models as SQL SELECT statements
Modular transformation logic
Version-controlled transformations
Testing and documentation
Dependency management

Example dbt Model:

-- models/marts/core/dim_customers.sql
{{
    config(
        materialized='table',
        sort='customer_id',
        dist='customer_id'
    )
}}

WITH customers AS (
    SELECT * FROM {{ ref('stg_customers') }}
),

orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
),

customer_orders AS (
    SELECT
        customer_id,
        MIN(order_date) AS first_order_date,
        MAX(order_date) AS most_recent_order_date,
        COUNT(order_id) AS number_of_orders,
        SUM(amount) AS lifetime_value
    FROM orders
    GROUP BY customer_id
),

final AS (
    SELECT
        customers.customer_id,
        customers.first_name,
        customers.last_name,
        customers.email,
        customers.created_at,
        customer_orders.first_order_date,
        customer_orders.most_recent_order_date,
        customer_orders.number_of_orders,
        customer_orders.lifetime_value
    FROM customers
    LEFT JOIN customer_orders USING (customer_id)
)

SELECT * FROM final

dbt Project Structure:

dbt_project/
├── dbt_project.yml          # Project configuration
├── packages.yml             # External dependencies
├── profiles.yml             # Connection profiles
├── README.md                # Project documentation
├── analysis/                # Ad-hoc analytical queries
├── data/                    # Seed data files
├── macros/                  # Reusable SQL functions
├── models/                  # SQL transformation models
│   ├── marts/               # Business-layer models
│   │   ├── core/
│   │   │   ├── dim_customers.sql
│   │   │   ├── dim_products.sql
│   │   │   ├── fct_orders.sql
│   │   │   └── schema.yml
│   │   └── marketing/
│   │       ├── customer_segmentation.sql
│   │       └── schema.yml
│   └── staging/             # Source-aligned models
│       ├── stg_customers.sql
│       ├── stg_orders.sql
│       ├── stg_products.sql
│       └── schema.yml
├── snapshots/               # Slowly changing dimension logic
└── tests/                   # Custom data tests

dbt Best Practices:

Follow a consistent naming convention
Implement a layered architecture
Write modular, reusable models
Document models and columns
Test critical assumptions

Incremental Processing

Efficiently handling growing datasets:

Incremental Load Patterns:

Timestamp-based incremental loads
Change data capture (CDC)
Slowly changing dimensions (SCD)
Merge operations
Partitioning strategies

Example Incremental dbt Model:

-- models/events/incremental_events.sql
{{
    config(
        materialized='incremental',
        unique_key='event_id',
        incremental_strategy='merge',
        partition_by={
            'field': 'event_date',
            'data_type': 'date'
        }
    )
}}

WITH source_data AS (
    SELECT
        event_id,
        event_type,
        user_id,
        event_timestamp,
        DATE(event_timestamp) AS event_date,
        payload
    FROM {{ source('events', 'raw_events') }}
    
    {% if is_incremental() %}
        -- Only process new or updated records when running incrementally
        WHERE event_timestamp > (
            SELECT MAX(event_timestamp) FROM {{ this }}
        )
    {% endif %}
)

SELECT
    event_id,
    event_type,
    user_id,
    event_timestamp,
    event_date,
    payload,
    {{ current_timestamp() }} AS processed_at
FROM source_data

Incremental Processing Challenges:

Handling late-arriving data
Managing schema evolution
Ensuring idempotent operations
Tracking processing metadata
Optimizing merge operations

Data Storage and Access Patterns

Data Warehouse Design

Structuring data for analytical workloads:

Schema Design Approaches:

Star schema
Snowflake schema
Data vault
One Big Table (OBT)
Hybrid approaches

Example Star Schema:

-- Fact table
CREATE TABLE fact_sales (
    sale_id INT PRIMARY KEY,
    date_id INT NOT NULL REFERENCES dim_date(date_id),
    product_id INT NOT NULL REFERENCES dim_product(product_id),
    customer_id INT NOT NULL REFERENCES dim_customer(customer_id),
    store_id INT NOT NULL REFERENCES dim_store(store_id),
    quantity INT NOT NULL,
    unit_price DECIMAL(10,2) NOT NULL,
    discount_amount DECIMAL(10,2) NOT NULL,
    sales_amount DECIMAL(10,2) NOT NULL,
    cost_amount DECIMAL(10,2) NOT NULL,
    profit_amount DECIMAL(10,2) NOT NULL
);

-- Dimension tables
CREATE TABLE dim_date (
    date_id INT PRIMARY KEY,
    date_actual DATE NOT NULL,
    day_of_week VARCHAR(10) NOT NULL,
    month_actual INT NOT NULL,
    month_name VARCHAR(10) NOT NULL,
    quarter_actual INT NOT NULL,
    year_actual INT NOT NULL,
    is_weekend BOOLEAN NOT NULL,
    is_holiday BOOLEAN NOT NULL
);

CREATE TABLE dim_product (
    product_id INT PRIMARY KEY,
    product_name VARCHAR(100) NOT NULL,
    category VARCHAR(50) NOT NULL,
    subcategory VARCHAR(50),
    unit_cost DECIMAL(10,2) NOT NULL,
    unit_price DECIMAL(10,2) NOT NULL
);

Partitioning and Clustering:

Time-based partitioning
Range partitioning
List partitioning
Hash partitioning
Clustering keys

Example BigQuery Partitioning and Clustering:

-- BigQuery partitioned and clustered table
CREATE OR REPLACE TABLE `project.dataset.fact_sales`
(
  sale_id STRING,
  sale_timestamp TIMESTAMP,
  customer_id STRING,
  product_id STRING,
  store_id STRING,
  quantity INT64,
  unit_price NUMERIC,
  sales_amount NUMERIC
)
PARTITION BY DATE(sale_timestamp)
CLUSTER BY store_id, product_id;

Data Lake Organization

Structuring raw and processed data:

Data Lake Zones:

Landing zone (raw data)
Bronze zone (validated raw data)
Silver zone (transformed/enriched data)
Gold zone (business-ready data)
Sandbox zone (exploration area)

Example Data Lake Structure:

data_lake/
├── landing/                 # Raw ingested data
│   ├── sales/
│   │   └── YYYY-MM-DD/      # Partitioned by ingestion date
│   ├── customers/
│   └── products/
├── bronze/                  # Validated raw data
│   ├── sales/
│   │   └── YYYY-MM-DD/      # Partitioned by ingestion date
│   ├── customers/
│   └── products/
├── silver/                  # Transformed data
│   ├── sales/
│   │   └── YYYY/MM/DD/      # Partitioned by business date
│   ├── customers/
│   └── products/
└── gold/                    # Business-ready data
    ├── analytics/
    │   ├── customer_360/
    │   └── sales_performance/
    └── reporting/
        ├── daily_sales_summary/
        └── monthly_kpis/

File Format Considerations:

Parquet for analytical workloads
Avro for schema evolution
ORC for columnar storage
JSON for flexibility
CSV for simplicity and compatibility

Streaming Data Processing

Stream Processing Patterns

Handling real-time data flows:

Event Streaming Architecture:

Producer/consumer model
Pub/sub messaging
Stream processing topologies
State management
Exactly-once processing

Common Stream Processing Operations:

Filtering and routing
Enrichment and transformation
Aggregation and windowing
Pattern detection
Joining streams

Stream Processing Technologies:

Apache Kafka Streams
Apache Flink
Apache Spark Structured Streaming
AWS Kinesis Data Analytics
Google Dataflow

Example Kafka Streams Application:

// Kafka Streams application for real-time sales analytics
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;

// Create topology
StreamsBuilder builder = new StreamsBuilder();

// Read from sales topic
KStream<String, Sale> salesStream = builder.stream(
    "sales-events", 
    Consumed.with(Serdes.String(), SaleSerdes.Sale())
);

// Calculate revenue by product category with 5-minute tumbling windows
salesStream
    .groupBy((key, sale) -> sale.getProductCategory())
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .aggregate(
        () -> 0.0,  // initializer
        (key, sale, total) -> total + sale.getAmount(),  // adder
        Materialized.with(Serdes.String(), Serdes.Double())
    )
    .toStream()
    .map((windowedKey, revenue) -> {
        String category = windowedKey.key();
        long windowStart = windowedKey.window().start();
        long windowEnd = windowedKey.window().end();
        return KeyValue.pair(
            category, 
            new CategoryRevenue(category, revenue, windowStart, windowEnd)
        );
    })
    .to(
        "category-revenue", 
        Produced.with(Serdes.String(), CategoryRevenueSerdes.CategoryRevenue())
    );

Stream Processing Best Practices:

Design for fault tolerance
Implement proper error handling
Consider state management carefully
Plan for data reprocessing
Monitor stream lag and throughput

Data Governance and Security

Data Governance

Establishing data management practices:

Data Governance Components:

Data cataloging and discovery
Metadata management
Data lineage tracking
Data quality monitoring
Policy enforcement

Data Catalog Implementation:

Document data sources and schemas
Track data transformations
Enable self-service discovery
Maintain business glossaries
Implement search capabilities

Data Lineage Tracking:

Capture source-to-target mappings
Visualize data flows
Track transformation logic
Enable impact analysis
Support compliance requirements

Data Security

Protecting sensitive data:

Security Best Practices:

Implement proper authentication and authorization
Encrypt data at rest and in transit
Apply column-level security
Implement row-level security
Maintain audit logs

Example Column-Level Security (Snowflake):

-- Create a masking policy for email addresses
CREATE OR REPLACE MASKING POLICY email_mask AS
  (val STRING) RETURNS STRING ->
    CASE
      WHEN CURRENT_ROLE() IN ('ANALYST', 'DATA_SCIENTIST') THEN 
        REGEXP_REPLACE(val, '^(.)(.*?)(@.*)', '$1***$3')
      WHEN CURRENT_ROLE() = 'DATA_ADMIN' THEN val
      ELSE '********'
    END;

-- Apply the masking policy to email columns
ALTER TABLE customers MODIFY COLUMN email
  SET MASKING POLICY email_mask;

ALTER TABLE employees MODIFY COLUMN email
  SET MASKING POLICY email_mask;

Data Privacy Techniques:

Data masking and tokenization
Dynamic data masking
Data anonymization
Differential privacy
Purpose-based access controls

Conclusion: Building Effective Data Pipelines

Data engineering is a critical discipline that enables organizations to transform raw data into valuable insights. By following the best practices outlined in this guide, you can build data pipelines that are scalable, reliable, and maintainable.

Key takeaways from this guide include:

Choose the Right Architecture: Select appropriate batch, streaming, or hybrid patterns based on your specific requirements
Prioritize Data Quality: Implement comprehensive testing and monitoring to ensure data reliability
Embrace Modern Tools: Leverage orchestration frameworks, transformation tools, and observability solutions
Design for Scale: Implement proper partitioning, incremental processing, and performance optimization
Establish Governance: Implement data cataloging, lineage tracking, and security controls

By applying these principles and leveraging the techniques discussed in this guide, you can build data infrastructure that delivers high-quality data to your organization’s analytical and operational workloads, enabling better decision-making and driving business value.

Andrew

Andrew is a visionary software engineer and DevOps expert with a proven track record of delivering cutting-edge solutions that drive innovation at Ataiva.com. As a leader on numerous high-profile projects, Andrew brings his exceptional technical expertise and collaborative leadership skills to the table, fostering a culture of agility and excellence within the team. With a passion for architecting scalable systems, automating workflows, and empowering teams, Andrew is a sought-after authority in the field of software development and DevOps.