etl-pipeline

ETL Pipeline for Construction Data

Overview

Based on DDC methodology (Chapter 4.2), this skill enables building automated data pipelines that extract information from various sources, transform it into useful formats, and load it into target systems or generate reports.

Book Reference: "ETL и автоматизация процессов" / "ETL and Process Automation"

"ETL: переход от ручного управления к автоматизации позволяет компаниям обрабатывать данные без постоянного человеческого вмешательства." — DDC Book, Chapter 4.2

ETL Components

┌─────────┐    ┌───────────┐    ┌────────┐
│ EXTRACT │ -> │ TRANSFORM │ -> │  LOAD  │
└─────────┘    └───────────┘    └────────┘
   │               │               │
   ▼               ▼               ▼
 Sources        Process         Outputs
 - PDF          - Clean         - Excel
 - Excel        - Validate      - PDF
 - CSV          - Calculate     - Database
 - BIM          - Merge         - API
 - API          - Aggregate     - Dashboard

Quick Start

import pandas as pd

# Simple ETL Pipeline
def simple_etl_pipeline(input_file, output_file):
    # EXTRACT
    df = pd.read_excel(input_file)

    # TRANSFORM
    df = df.dropna()  # Clean
    df['Total'] = df['Quantity'] * df['Unit_Price']  # Calculate
    summary = df.groupby('Category')['Total'].sum()  # Aggregate

    # LOAD
    summary.to_excel(output_file)
    return summary

# Run
result = simple_etl_pipeline("raw_data.xlsx", "processed_report.xlsx")

Extract: Data Sources

From Multiple Excel Files

import pandas as pd
from pathlib import Path

def extract_excel_files(folder_path, pattern="*.xlsx"):
    """Extract data from multiple Excel files"""
    files = Path(folder_path).glob(pattern)
    all_data = []

    for file in files:
        try:
            df = pd.read_excel(file)
            df['_source_file'] = file.name
            all_data.append(df)
            print(f"Extracted: {file.name}")
        except Exception as e:
            print(f"Error reading {file.name}: {e}")

    if all_data:
        return pd.concat(all_data, ignore_index=True)
    return pd.DataFrame()

# Usage
df = extract_excel_files("./project_data/")