Python Logging: Levels, Formatters, Handlers, Rotating Files, Structured Logging, and Every Pattern Data Engineers Need in Production

Your pipeline ran last night. Did it succeed? How many rows did it process? Did it skip any bad records? How long did each step take? Was there a warning about a slow query? Without logging, the answer to all of these is “I have no idea — let me re-run it and watch the terminal.”

print() statements are fine for quick debugging. But they are not logging. They cannot be filtered by severity. They cannot be routed to files, monitoring systems, or Slack channels. They cannot be turned on and off without editing code. They disappear the moment the terminal closes. They are the equivalent of writing notes on napkins — useful in the moment, lost immediately after.

Real-life analogy: Think of logging like the black box in an airplane. The plane records altitude, speed, engine temperature, and pilot commands continuously. If the flight is smooth, nobody looks at the data. But if something goes wrong, that black box is the first thing investigators open. Python’s logging module is your pipeline’s black box — it records everything that happens so you can reconstruct exactly what went wrong, when, and why.

Table of Contents

• Why Not Just Use print()?
• print() vs logging Side by Side
• The 5 Reasons to Use logging
• Getting Started with logging
• Basic Configuration
• Your First Logger
• Log Levels
• The 5 Standard Levels
• Choosing the Right Level
• Setting the Minimum Level
• Formatters — Controlling What Gets Logged
• Built-in Format Attributes
• Custom Formatters
• Handlers — Where Logs Go
• StreamHandler (Console)
• FileHandler (Log Files)
• Multiple Handlers (Console + File)
• RotatingFileHandler (Size-Based Rotation)
• TimedRotatingFileHandler (Date-Based Rotation)
• Named Loggers and Hierarchy
• Why Named Loggers
• Logger Hierarchy and Propagation
• Logger Configuration Pattern
• Logging in Functions and Classes
• Module-Level Logger Pattern
• Class Logger Pattern
• Logging Exceptions
• Structured Logging (JSON)
• Why Structured Logging
• JSON Formatter
• Adding Extra Context
• Configuration Methods
• basicConfig (Quick Start)
• Code-Based Configuration (Recommended)
• Dictionary Configuration
• Data Engineering Patterns
• Pipeline Run Logger
• Step Timer Logger
• Row Count and Metrics Logger
• Error Quarantine Logger
• Multi-Table Pipeline Logger
• Common Mistakes
• Interview Questions
• Wrapping Up

Why Not Just Use print()?

print() vs logging Side by Side

# ❌ The print() approach — fine for debugging, terrible for production
print("Pipeline started")
print(f"Loading table: customers")
print(f"Loaded 50000 rows")
print("WARNING: 12 rows had null customer_id")
print("ERROR: Connection timeout on retry 3")
print("Pipeline finished")

# ✅ The logging approach — production-ready
import logging
logger = logging.getLogger("pipeline")

logger.info("Pipeline started")
logger.info("Loading table: customers")
logger.info("Loaded 50,000 rows")
logger.warning("12 rows had null customer_id")
logger.error("Connection timeout on retry 3")
logger.info("Pipeline finished")

# Output with logging:
# 2026-07-08 14:30:00 | INFO    | pipeline | Pipeline started
# 2026-07-08 14:30:01 | INFO    | pipeline | Loading table: customers
# 2026-07-08 14:30:05 | INFO    | pipeline | Loaded 50,000 rows
# 2026-07-08 14:30:05 | WARNING | pipeline | 12 rows had null customer_id
# 2026-07-08 14:30:15 | ERROR   | pipeline | Connection timeout on retry 3
# 2026-07-08 14:30:15 | INFO    | pipeline | Pipeline finished

The 5 Reasons to Use logging

1. Severity levels — filter by importance (show only warnings and errors in production, show everything in dev).
2. Timestamps — know exactly when each event happened without manually adding datetime.now() to every print.
3. Multiple destinations — send logs to console AND a file AND a monitoring system simultaneously.
4. No code changes to control output — change the log level from INFO to DEBUG without editing a single line of code.
5. Structured data — output as JSON for log aggregation tools (Splunk, ELK, CloudWatch, Azure Monitor).

Getting Started with logging

Basic Configuration

import logging

# Quickest possible setup — one line
logging.basicConfig(level=logging.INFO)
logging.info("Pipeline started")
# INFO:root:Pipeline started

# Better — with formatting and file output
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logging.info("Pipeline started")
# 2026-07-08 14:30:00 | INFO     | root | Pipeline started

# Write to a file instead of console
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    filename="pipeline.log",
    filemode="a",   # "a" = append, "w" = overwrite each run
)

# IMPORTANT: basicConfig only works ONCE. If you call it twice,
# the second call is silently ignored. For more control, use handlers.

Your First Logger

import logging

# Create a named logger (NOT the root logger)
logger = logging.getLogger("my_pipeline")
logger.setLevel(logging.DEBUG)

# Create a console handler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)   # Console shows INFO and above

# Create a formatter
formatter = logging.Formatter(
    "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
console_handler.setFormatter(formatter)

# Add handler to logger
logger.addHandler(console_handler)

# Use it
logger.debug("This is a debug message")     # NOT shown (console level is INFO)
logger.info("Pipeline started")              # Shown
logger.warning("Slow query detected")        # Shown
logger.error("Connection failed")            # Shown

# Output:
# 2026-07-08 14:30:00 | INFO     | my_pipeline | Pipeline started
# 2026-07-08 14:30:01 | WARNING  | my_pipeline | Slow query detected
# 2026-07-08 14:30:02 | ERROR    | my_pipeline | Connection failed

Log Levels

The 5 Standard Levels

Level	Numeric	When to Use	Example
DEBUG	10	Detailed diagnostic info — only during development	Variable values, SQL queries, row counts per step
INFO	20	Normal operation — confirmation things are working	“Pipeline started”, “Loaded 50K rows”, “Step completed”
WARNING	30	Something unexpected but not broken — investigate soon	“12 null IDs skipped”, “Retry 2 of 3”, “Slow query 8.2s”
ERROR	40	Something failed — a specific operation could not complete	“Connection timeout”, “File not found”, “API returned 500”
CRITICAL	50	System is broken — the entire pipeline cannot continue	“Database is down”, “Config file missing”, “Disk full”

Real-life analogy: Think of log levels like a hospital triage system. DEBUG is the nurse taking detailed vitals on every patient. INFO is the admissions desk confirming patients are checked in. WARNING is a patient whose temperature is slightly elevated — watch them. ERROR is a patient who needs immediate treatment — something is broken. CRITICAL is a code blue — the entire floor needs to respond.

Choosing the Right Level

import logging
logger = logging.getLogger("etl")

# DEBUG — detailed internals (off in production)
logger.debug(f"SQL query: SELECT * FROM customers WHERE active = 1")
logger.debug(f"Raw API response: {response.text[:200]}")
logger.debug(f"Column types: {df.dtypes.to_dict()}")

# INFO — normal milestones (the heartbeat of your pipeline)
logger.info("Pipeline started: daily_customer_load")
logger.info(f"Source: customers table ({row_count:,} rows)")
logger.info(f"Loaded to silver.customers in {duration:.1f}s")
logger.info("Pipeline completed successfully")

# WARNING — something is off but pipeline continues
logger.warning(f"Skipped {null_count} rows with null customer_id")
logger.warning(f"Retry 2/3: API returned 503")
logger.warning(f"Query took {elapsed:.1f}s (threshold: 5.0s)")

# ERROR — something failed (but pipeline may continue for other tables)
logger.error(f"Failed to load table 'orders': {e}")
logger.error(f"API call failed after 3 retries: {url}")

# CRITICAL — pipeline cannot continue at all
logger.critical("Database connection pool exhausted — shutting down")
logger.critical(f"Config file not found: {config_path}")

Setting the Minimum Level

# Setting level = "show this level and everything above"
logger.setLevel(logging.DEBUG)    # Shows: DEBUG, INFO, WARNING, ERROR, CRITICAL
logger.setLevel(logging.INFO)     # Shows: INFO, WARNING, ERROR, CRITICAL
logger.setLevel(logging.WARNING)  # Shows: WARNING, ERROR, CRITICAL
logger.setLevel(logging.ERROR)    # Shows: ERROR, CRITICAL

# Common pattern: DEBUG in dev, INFO in production
import os
env = os.getenv("ENVIRONMENT", "production")
log_level = logging.DEBUG if env == "development" else logging.INFO
logger.setLevel(log_level)

# You can also set levels per HANDLER (more flexible)
# Logger level = gate 1, Handler level = gate 2
# A message must pass BOTH gates to be output

Formatters — Controlling What Gets Logged

Built-in Format Attributes

Attribute	Format	Output Example
Timestamp	%(asctime)s	2026-07-08 14:30:00
Level name	%(levelname)s	INFO, WARNING, ERROR
Level (padded)	%(levelname)-8s	INFO    , WARNING
Logger name	%(name)s	my_pipeline, etl.customers
Message	%(message)s	Loaded 50,000 rows
Module name	%(module)s	pipeline (filename without .py)
Function name	%(funcName)s	load_customers
Line number	%(lineno)d	42
Process ID	%(process)d	12345
Thread name	%(threadName)s	MainThread

Custom Formatters

import logging

# Simple format — good for development
dev_format = logging.Formatter(
    "%(levelname)-8s | %(message)s"
)
# INFO     | Pipeline started

# Standard format — good for production log files
prod_format = logging.Formatter(
    "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
# 2026-07-08 14:30:00 | INFO     | my_pipeline | Pipeline started

# Detailed format — good for debugging
debug_format = logging.Formatter(
    "%(asctime)s | %(levelname)-8s | %(name)s | %(funcName)s:%(lineno)d | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
# 2026-07-08 14:30:00 | INFO     | my_pipeline | load_customers:42 | Pipeline started

# Different formatters for different handlers:
console_handler.setFormatter(dev_format)     # Short on screen
file_handler.setFormatter(prod_format)       # Detailed in file

Handlers — Where Logs Go

StreamHandler (Console)

import logging

logger = logging.getLogger("pipeline")
logger.setLevel(logging.DEBUG)

# Console handler — prints to terminal
console = logging.StreamHandler()
console.setLevel(logging.INFO)    # Console only shows INFO+
console.setFormatter(logging.Formatter(
    "%(asctime)s | %(levelname)-8s | %(message)s",
    datefmt="%H:%M:%S"
))
logger.addHandler(console)

logger.info("Pipeline started")
# 14:30:00 | INFO     | Pipeline started

FileHandler (Log Files)

import logging

logger = logging.getLogger("pipeline")
logger.setLevel(logging.DEBUG)

# File handler — writes to a log file
file_handler = logging.FileHandler("pipeline.log", mode="a", encoding="utf-8")
file_handler.setLevel(logging.DEBUG)   # File captures EVERYTHING
file_handler.setFormatter(logging.Formatter(
    "%(asctime)s | %(levelname)-8s | %(name)s | %(funcName)s:%(lineno)d | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
))
logger.addHandler(file_handler)

# mode="a" → append (keeps history)
# mode="w" → overwrite each run (fresh log per run)

Multiple Handlers (Console + File)

import logging

def setup_logger(name, log_file="pipeline.log", level=logging.DEBUG):
    """Create a logger with both console and file output."""
    logger = logging.getLogger(name)
    logger.setLevel(level)

    # Prevent duplicate handlers if called multiple times
    if logger.handlers:
        return logger

    # Console handler — INFO and above (clean output)
    console = logging.StreamHandler()
    console.setLevel(logging.INFO)
    console.setFormatter(logging.Formatter(
        "%(asctime)s | %(levelname)-8s | %(message)s",
        datefmt="%H:%M:%S"
    ))

    # File handler — DEBUG and above (full detail)
    file_h = logging.FileHandler(log_file, encoding="utf-8")
    file_h.setLevel(logging.DEBUG)
    file_h.setFormatter(logging.Formatter(
        "%(asctime)s | %(levelname)-8s | %(name)s | %(funcName)s:%(lineno)d | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    ))

    logger.addHandler(console)
    logger.addHandler(file_h)
    return logger

# Usage
logger = setup_logger("etl_pipeline")
logger.debug("SQL: SELECT * FROM customers")   # File only
logger.info("Loaded 50,000 rows")              # Console + File
logger.warning("12 null IDs skipped")          # Console + File

Real-life analogy: Handlers are like mail delivery routes. A StreamHandler is like a PA announcement — everyone in the building hears it. A FileHandler is like a registered letter — it is recorded and archived. A pipeline with both is like a hospital that announces emergencies on the PA system and records them in the patient file. The PA only announces critical events (INFO+), but the file captures every detail (DEBUG+).

RotatingFileHandler (Size-Based Rotation)

from logging.handlers import RotatingFileHandler

# Rotate when file reaches 5 MB, keep 3 backup files
rotating_handler = RotatingFileHandler(
    "pipeline.log",
    maxBytes=5 * 1024 * 1024,    # 5 MB
    backupCount=3,                # Keep 3 old files
    encoding="utf-8"
)
# Files created: pipeline.log, pipeline.log.1, pipeline.log.2, pipeline.log.3
# When pipeline.log hits 5 MB:
#   pipeline.log.2 → pipeline.log.3 (oldest is deleted if backupCount exceeded)
#   pipeline.log.1 → pipeline.log.2
#   pipeline.log   → pipeline.log.1
#   New pipeline.log is created (empty)

rotating_handler.setFormatter(logging.Formatter(
    "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
))
logger.addHandler(rotating_handler)

TimedRotatingFileHandler (Date-Based Rotation)

from logging.handlers import TimedRotatingFileHandler

# Rotate at midnight, keep 30 days of logs
timed_handler = TimedRotatingFileHandler(
    "pipeline.log",
    when="midnight",       # Rotate at midnight
    interval=1,            # Every 1 day
    backupCount=30,        # Keep 30 days
    encoding="utf-8"
)
timed_handler.suffix = "%Y-%m-%d"   # Backup filenames: pipeline.log.2026-07-07
# Files: pipeline.log (current), pipeline.log.2026-07-07, pipeline.log.2026-07-06, ...

# Other "when" options:
# "S" = seconds, "M" = minutes, "H" = hours
# "D" = days, "midnight" = midnight, "W0"-"W6" = weekday (Mon-Sun)

Named Loggers and Hierarchy

Why Named Loggers

# Named loggers let you control logging per module/component
# Instead of one global logger, each module gets its own

# pipeline/extract.py
logger = logging.getLogger("pipeline.extract")
logger.info("Extracting from source")

# pipeline/transform.py
logger = logging.getLogger("pipeline.transform")
logger.info("Transforming data")

# pipeline/load.py
logger = logging.getLogger("pipeline.load")
logger.info("Loading to warehouse")

# Output shows which module logged the message:
# 2026-07-08 14:30:00 | INFO | pipeline.extract   | Extracting from source
# 2026-07-08 14:30:05 | INFO | pipeline.transform | Transforming data
# 2026-07-08 14:30:10 | INFO | pipeline.load      | Loading to warehouse

# You can also change levels per module:
logging.getLogger("pipeline.extract").setLevel(logging.DEBUG)    # Verbose
logging.getLogger("pipeline.load").setLevel(logging.WARNING)     # Quiet

Logger Hierarchy and Propagation

# Loggers form a tree based on dot-separated names
# "pipeline"            → parent
# "pipeline.extract"    → child
# "pipeline.transform"  → child

# By default, log messages PROPAGATE UP to parent loggers
# If "pipeline" has a handler, it receives messages from ALL children

# Set up the PARENT logger with handlers
parent = logging.getLogger("pipeline")
parent.setLevel(logging.DEBUG)
parent.addHandler(console_handler)
parent.addHandler(file_handler)

# Child loggers inherit the parent's handlers — no setup needed
extract_logger = logging.getLogger("pipeline.extract")
extract_logger.info("Reading from S3")
# This message goes to console_handler AND file_handler
# (inherited from "pipeline")

# To STOP propagation (child has its own handlers):
extract_logger.propagate = False   # Messages stay at this level

Logger Configuration Pattern

import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path

def configure_logging(
    name="pipeline",
    log_dir="logs",
    console_level=logging.INFO,
    file_level=logging.DEBUG,
    max_bytes=10 * 1024 * 1024,   # 10 MB
    backup_count=5
):
    """One-time logging setup for a pipeline. Call once at the start."""
    # Create log directory
    Path(log_dir).mkdir(parents=True, exist_ok=True)

    # Get or create logger
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)   # Capture everything — handlers filter

    # Skip if already configured
    if logger.handlers:
        return logger

    # Console handler
    console = logging.StreamHandler()
    console.setLevel(console_level)
    console.setFormatter(logging.Formatter(
        "%(asctime)s | %(levelname)-8s | %(message)s",
        datefmt="%H:%M:%S"
    ))

    # Rotating file handler
    log_file = Path(log_dir) / f"{name}.log"
    file_h = RotatingFileHandler(
        log_file, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8"
    )
    file_h.setLevel(file_level)
    file_h.setFormatter(logging.Formatter(
        "%(asctime)s | %(levelname)-8s | %(name)s | %(funcName)s:%(lineno)d | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    ))

    logger.addHandler(console)
    logger.addHandler(file_h)

    return logger

# Usage — call ONCE at pipeline start
logger = configure_logging("etl_pipeline", log_dir="logs")
logger.info("Pipeline initialized")

Logging in Functions and Classes

Module-Level Logger Pattern

# THE standard pattern — one logger per module, defined at the top
# Python convention: use __name__ as the logger name
# __name__ automatically becomes the module's dotted path

# file: pipeline/extract.py
import logging

logger = logging.getLogger(__name__)   # logger name = "pipeline.extract"

def extract_from_s3(bucket, key):
    logger.info(f"Extracting: s3://{bucket}/{key}")
    try:
        data = s3_client.get_object(Bucket=bucket, Key=key)
        logger.debug(f"Response metadata: {data['ResponseMetadata']}")
        return data["Body"].read()
    except Exception as e:
        logger.error(f"Failed to extract s3://{bucket}/{key}: {e}")
        raise

def extract_from_api(url, params=None):
    logger.info(f"Extracting from API: {url}")
    response = requests.get(url, params=params, timeout=30)
    logger.debug(f"Status: {response.status_code}, Size: {len(response.content):,} bytes")
    response.raise_for_status()
    return response.json()

Class Logger Pattern

import logging

class DataPipeline:
    def __init__(self, name, config):
        self.name = name
        self.config = config
        self.logger = logging.getLogger(f"pipeline.{name}")

    def extract(self):
        self.logger.info(f"Extracting from {self.config['source']}")
        # ... extraction logic
        self.logger.info(f"Extracted {row_count:,} rows")

    def transform(self, df):
        self.logger.info(f"Transforming {len(df):,} rows")
        # ... transformation logic
        self.logger.info(f"Transformation complete: {len(result):,} rows")

    def load(self, df, target):
        self.logger.info(f"Loading {len(df):,} rows to {target}")
        # ... load logic
        self.logger.info(f"Load complete")

# Each pipeline instance gets its own named logger:
customers = DataPipeline("customers", customer_config)
# Logger: "pipeline.customers"

orders = DataPipeline("orders", order_config)
# Logger: "pipeline.orders"

Logging Exceptions

import logging

logger = logging.getLogger("pipeline")

# logger.exception() — logs ERROR level WITH the full traceback
try:
    result = 10 / 0
except ZeroDivisionError:
    logger.exception("Calculation failed")
    # ERROR | Calculation failed
    # Traceback (most recent call last):
    #   File "pipeline.py", line 5, in <module>
    #     result = 10 / 0
    # ZeroDivisionError: division by zero

# logger.error() with exc_info=True — same result, but you control the level
try:
    data = json.load(open("broken.json"))
except json.JSONDecodeError:
    logger.error("Failed to parse JSON", exc_info=True)  # Includes traceback

# logger.warning() with exc_info — for non-fatal errors you want to track
try:
    cache = load_cache()
except FileNotFoundError:
    logger.warning("Cache not found — starting fresh", exc_info=True)
    cache = {}

# NEVER use logger.exception() outside an except block — it will log "None" as traceback

Structured Logging (JSON)

Why Structured Logging

Plain text logs are human-readable but hard to search and filter at scale. When your logs go to CloudWatch, Splunk, ELK Stack, or Azure Monitor, structured JSON logs let you query by any field: level="ERROR" AND table="customers" AND duration>5.0. This is impossible with plain text unless you parse every log line with regex.

JSON Formatter

import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    """Format log records as JSON — one JSON object per line."""

    def format(self, record):
        log_data = {
            "timestamp": datetime.fromtimestamp(record.created).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }

        # Include exception info if present
        if record.exc_info and record.exc_info[0]:
            log_data["exception"] = self.formatException(record.exc_info)

        # Include any extra fields
        if hasattr(record, "extra_data"):
            log_data.update(record.extra_data)

        return json.dumps(log_data)

# Usage
json_handler = logging.FileHandler("pipeline.json.log", encoding="utf-8")
json_handler.setFormatter(JSONFormatter())
logger.addHandler(json_handler)

logger.info("Loaded customers table")
# {"timestamp": "2026-07-08T14:30:00", "level": "INFO", "logger": "pipeline",
#  "message": "Loaded customers table", "module": "main", "function": "load", "line": 42}

Adding Extra Context

# Method 1: Using the extra parameter
logger.info("Table loaded", extra={"extra_data": {
    "table": "customers",
    "rows": 50000,
    "duration_seconds": 3.2,
    "source": "azure_sql"
}})
# {"timestamp": "...", "level": "INFO", "message": "Table loaded",
#  "table": "customers", "rows": 50000, "duration_seconds": 3.2, "source": "azure_sql"}

# Method 2: Using LoggerAdapter for persistent context
class PipelineAdapter(logging.LoggerAdapter):
    def process(self, msg, kwargs):
        # Add pipeline_name and run_id to every log message
        extra = kwargs.get("extra", {})
        extra["extra_data"] = {**self.extra, **extra.get("extra_data", {})}
        kwargs["extra"] = extra
        return msg, kwargs

adapter = PipelineAdapter(logger, {
    "pipeline": "daily_customer_load",
    "run_id": "run_20260708_143000"
})
adapter.info("Pipeline started")
# Every message now includes pipeline name and run_id automatically

Configuration Methods

basicConfig (Quick Start)

# Simplest setup — one line, works for scripts
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

# Limitations:
# - Only works ONCE (second call is ignored)
# - Cannot easily add multiple handlers
# - Cannot configure per-module levels
# Best for: quick scripts, notebooks, one-off analyses

Code-Based Configuration (Recommended)

# The setup_logger / configure_logging function shown earlier
# Full control, reusable, easy to understand
# Best for: production pipelines, applications

logger = configure_logging("etl_pipeline", log_dir="logs")

Dictionary Configuration

import logging.config

LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "standard": {
            "format": "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
            "datefmt": "%Y-%m-%d %H:%M:%S"
        },
        "detailed": {
            "format": "%(asctime)s | %(levelname)-8s | %(name)s | %(funcName)s:%(lineno)d | %(message)s",
            "datefmt": "%Y-%m-%d %H:%M:%S"
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "standard",
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "DEBUG",
            "formatter": "detailed",
            "filename": "logs/pipeline.log",
            "maxBytes": 10485760,   # 10 MB
            "backupCount": 5,
            "encoding": "utf-8"
        }
    },
    "loggers": {
        "pipeline": {
            "level": "DEBUG",
            "handlers": ["console", "file"],
        },
        "pipeline.extract": {
            "level": "DEBUG",     # Verbose for extraction
        },
        "pipeline.load": {
            "level": "INFO",      # Quiet for loading
        }
    }
}

# Apply the config
logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger("pipeline")

# Best for: large projects, config stored in JSON/YAML files

Data Engineering Patterns

Pipeline Run Logger

import logging
from datetime import datetime

def run_pipeline(pipeline_name, tables):
    """Run a pipeline with structured logging."""
    logger = logging.getLogger(f"pipeline.{pipeline_name}")
    run_id = datetime.now().strftime("%Y%m%d_%H%M%S")

    logger.info(f"[{run_id}] Pipeline '{pipeline_name}' STARTED — {len(tables)} tables")

    results = {"success": [], "failed": []}

    for table in tables:
        try:
            logger.info(f"[{run_id}] Loading: {table}")
            row_count = load_table(table)
            logger.info(f"[{run_id}] Loaded: {table} ({row_count:,} rows)")
            results["success"].append(table)
        except Exception as e:
            logger.error(f"[{run_id}] FAILED: {table} — {e}")
            results["failed"].append(table)

    if results["failed"]:
        logger.warning(
            f"[{run_id}] Pipeline completed with errors: "
            f"{len(results['success'])} succeeded, {len(results['failed'])} failed"
        )
    else:
        logger.info(f"[{run_id}] Pipeline completed: all {len(tables)} tables loaded")

    return results

Step Timer Logger

import logging
import time
from contextlib import contextmanager

logger = logging.getLogger("pipeline")

@contextmanager
def log_step(step_name):
    """Context manager that logs step duration."""
    logger.info(f"Step '{step_name}' started")
    start = time.time()
    try:
        yield
    except Exception as e:
        duration = time.time() - start
        logger.error(f"Step '{step_name}' FAILED after {duration:.1f}s — {e}")
        raise
    else:
        duration = time.time() - start
        level = logging.WARNING if duration > 60 else logging.INFO
        logger.log(level, f"Step '{step_name}' completed in {duration:.1f}s")

# Usage — automatically logs duration and warns if slow
with log_step("extract_customers"):
    df = extract_from_source("customers")

with log_step("transform_customers"):
    df = apply_transformations(df)

with log_step("load_customers"):
    load_to_warehouse(df, "silver.customers")

# Output:
# 14:30:00 | INFO     | Step 'extract_customers' started
# 14:30:03 | INFO     | Step 'extract_customers' completed in 3.2s
# 14:30:03 | INFO     | Step 'transform_customers' started
# 14:30:05 | INFO     | Step 'transform_customers' completed in 1.8s
# 14:30:05 | INFO     | Step 'load_customers' started
# 14:31:10 | WARNING  | Step 'load_customers' completed in 65.4s

Row Count and Metrics Logger

import logging

logger = logging.getLogger("pipeline.metrics")

def log_dataframe_stats(df, stage_name):
    """Log key metrics about a DataFrame at each pipeline stage."""
    logger.info(f"[{stage_name}] Rows: {len(df):,}")
    logger.info(f"[{stage_name}] Columns: {list(df.columns)}")

    null_counts = df.isnull().sum()
    if null_counts.any():
        nulls = null_counts[null_counts > 0].to_dict()
        logger.warning(f"[{stage_name}] Null values: {nulls}")

    dupes = df.duplicated().sum()
    if dupes > 0:
        logger.warning(f"[{stage_name}] Duplicate rows: {dupes:,}")

    logger.debug(f"[{stage_name}] Dtypes: {df.dtypes.to_dict()}")
    logger.debug(f"[{stage_name}] Memory: {df.memory_usage(deep=True).sum() / 1024 / 1024:.1f} MB")

# Usage at each stage
df_raw = extract("customers")
log_dataframe_stats(df_raw, "bronze")

df_clean = transform(df_raw)
log_dataframe_stats(df_clean, "silver")

# Output:
# [bronze] Rows: 52,340
# [bronze] Columns: ['id', 'name', 'email', 'created_at']
# [bronze] Null values: {'email': 47, 'created_at': 3}
# [silver] Rows: 52,290
# [silver] Columns: ['id', 'name', 'email', 'created_at', 'load_date']

Error Quarantine Logger

import logging
import json

# Separate logger for quarantined records — goes to its own file
quarantine_logger = logging.getLogger("pipeline.quarantine")
quarantine_handler = logging.FileHandler("quarantine.log", encoding="utf-8")
quarantine_handler.setFormatter(logging.Formatter(
    "%(asctime)s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
))
quarantine_logger.addHandler(quarantine_handler)

def process_with_quarantine(records):
    """Process records, quarantine bad ones with full context."""
    good, bad = [], []

    for i, record in enumerate(records):
        try:
            validated = validate_and_transform(record)
            good.append(validated)
        except Exception as e:
            bad.append(record)
            quarantine_logger.error(json.dumps({
                "row_index": i,
                "error": str(e),
                "error_type": type(e).__name__,
                "record": record
            }))

    logger.info(f"Processed: {len(good)} good, {len(bad)} quarantined")
    return good

Multi-Table Pipeline Logger

import logging
from datetime import datetime

def run_multi_table_pipeline(config):
    """Full pipeline with comprehensive logging."""
    logger = logging.getLogger("pipeline")
    run_id = datetime.now().strftime("%Y%m%d_%H%M%S")
    tables = config["tables"]

    logger.info("=" * 60)
    logger.info(f"PIPELINE START | run_id={run_id} | tables={len(tables)}")
    logger.info("=" * 60)

    summary = {"loaded": 0, "failed": 0, "total_rows": 0, "errors": []}
    start_time = time.time()

    for table_name in tables:
        table_start = time.time()
        try:
            logger.info(f"[{table_name}] Extracting...")
            df = extract(table_name, config)
            logger.info(f"[{table_name}] Extracted {len(df):,} rows")

            logger.info(f"[{table_name}] Transforming...")
            df = transform(df, config)

            logger.info(f"[{table_name}] Loading to warehouse...")
            load(df, f"silver.{table_name}")

            duration = time.time() - table_start
            logger.info(f"[{table_name}] ✓ Complete ({len(df):,} rows in {duration:.1f}s)")
            summary["loaded"] += 1
            summary["total_rows"] += len(df)

        except Exception as e:
            duration = time.time() - table_start
            logger.error(f"[{table_name}] ✗ Failed after {duration:.1f}s: {e}")
            summary["failed"] += 1
            summary["errors"].append({"table": table_name, "error": str(e)})

    total_duration = time.time() - start_time
    logger.info("=" * 60)
    logger.info(
        f"PIPELINE END | run_id={run_id} | "
        f"loaded={summary['loaded']} | failed={summary['failed']} | "
        f"rows={summary['total_rows']:,} | duration={total_duration:.1f}s"
    )
    if summary["errors"]:
        for err in summary["errors"]:
            logger.error(f"  FAILED: {err['table']} — {err['error']}")
    logger.info("=" * 60)

    return summary

Common Mistakes

1. Using print() in production code — print cannot be filtered, routed, or structured. Switch to logging for anything that runs unattended or in production.
2. Calling basicConfig() multiple times — only the first call takes effect. Subsequent calls are silently ignored. Use named loggers with explicit handlers instead.
3. Not using __name__ for logger names — hardcoding logger names makes it impossible to trace which module generated a log message. logging.getLogger(__name__) gives each module a unique, traceable name.
4. Logging sensitive data — never log passwords, API keys, connection strings, or PII (emails, SSNs). Redact or mask sensitive fields: logger.info(f"Connecting to {host} as {user}") — never include the password.
5. Not logging the exception traceback — logger.error(f"Failed: {e}") only logs the message. Use logger.exception("Failed") or logger.error("Failed", exc_info=True) to include the full traceback.
6. Creating duplicate handlers — calling addHandler() multiple times adds duplicate handlers, causing the same message to appear twice. Always check if logger.handlers: before adding.
7. Setting the logger level too high — setting the logger to WARNING means you lose all INFO messages. Set the logger to DEBUG and control verbosity at the handler level (console=INFO, file=DEBUG).

Interview Questions

Q: Why use the logging module instead of print()? A: The logging module provides severity levels (DEBUG through CRITICAL) so you can filter output without changing code. It supports multiple destinations — console, files, and monitoring systems simultaneously. It adds timestamps and context automatically. And it can be configured at runtime without editing code, making it essential for production systems where you cannot watch the terminal.

Q: What are the five standard log levels and when do you use each? A: DEBUG (10) for detailed diagnostics during development. INFO (20) for normal operation milestones. WARNING (30) for unexpected but non-breaking situations. ERROR (40) for failed operations. CRITICAL (50) for system-breaking failures. Set production loggers to INFO — you get operational visibility without diagnostic noise.

Q: What is the difference between a logger, a handler, and a formatter? A: The logger decides IF a message should be processed (based on level). The handler decides WHERE it goes (console, file, network). The formatter decides HOW it looks (timestamp format, fields included). One logger can have multiple handlers, each with its own formatter and level — so the console shows short INFO+ messages while the file captures detailed DEBUG+ messages.

Q: How do you avoid duplicate log messages? A: Check if logger.handlers: before adding handlers. Use propagate = False on child loggers if they have their own handlers and you do not want messages sent to parent loggers. In dictionary config, set disable_existing_loggers: False to prevent loggers created before config from being silenced.

Q: What is a RotatingFileHandler and why is it important? A: A RotatingFileHandler automatically creates a new log file when the current one reaches a size limit (e.g., 10 MB), keeping a configurable number of old files. This prevents log files from growing indefinitely and consuming all disk space — critical for long-running pipelines and services.

Q: How would you set up logging for a production data pipeline? A: Use named loggers per module (getLogger(__name__)), a console handler at INFO level for operational visibility, a rotating file handler at DEBUG level for full diagnostics, structured JSON formatting for log aggregation tools, and a separate quarantine logger for bad records. Configure once at pipeline start and use log_step context managers to automatically time each stage.

Wrapping Up

Logging is the observability layer of your pipeline. Without it, you are flying blind. The logging module gives you severity levels for filtering, handlers for routing to multiple destinations, formatters for controlling output, and named loggers for per-module control. The key patterns for data engineering are: configure once at startup, use named loggers per module, time every step, log row counts at every stage, quarantine bad records to a separate log, and use structured JSON when your logs feed into a monitoring system.

The goal is simple: when something goes wrong at 3 AM, your logs should tell you exactly what happened, when, where in the code, and what data was involved — without re-running anything.

Previous in this series: Error Handling — try/except/finally, Custom Exceptions

Next in this series: OOP — Classes, __init__, self, Methods, and Properties

---

Naveen Vuppula is a Senior Data Engineering Consultant and app developer based in Ontario, Canada. He writes about Python, SQL, AWS, Azure, and everything data engineering at DriveDataScience.com.