Real-Time CSV Stream Processor

Problem Statement

Many data ingestion pipelines rely on log files, transaction dumps, or sensor feeds that append records incrementally to CSV files. Standard ETL tools often require reprocessing the entire CSV file from scratch on every run or rely on heavyweight stream-processing frameworks like Apache Flink or Spark. This creates unnecessary compute overhead, slow feedback loops, and infrastructure complexity for simple streaming tasks.

This script provides a lightweight, pure-Python stream processor that tails a growing CSV file, dynamically executes custom transformation logic (via pandas), and streams the resulting records into structured output destinations (JSON Lines or CSV sinks). It handles incremental tailing via byte-offset tracking, partial line writes, file truncations, and smooth graceful shutdowns.

How It Works

The Real-Time CSV Stream Processor uses a byte-offset tracking engine to poll the target CSV file without loading the entire dataset into memory.

1. Header Discovery & Initialization: Upon starting, the processor opens the target file, extracts the CSV header, and initializes the byte offset.
2. Byte-Buffered Tailing: During each polling cycle, the processor reads new bytes appended after the recorded offset. To prevent reading partial lines, it slices the buffer up to the final newline character (\n) and saves the remaining unparsed bytes for the next cycle.
3. Dynamic Rule Loading: If a custom --transform script is provided, the script dynamically imports it using Python's importlib.util module and passes the incoming batch as a pandas.DataFrame to the module's transform(df) function.
4. Structured Sink Writing: Transformed records are serialized and appended directly to the target output sink. JSON Lines (.jsonl) writes one JSON record per line; CSV output appends rows while managing header headers automatically.

       +-----------------------+
       |   Growing Input CSV   |
       +-----------+-----------+
                   |
         [ Byte Offset Tail ]
                   |
                   v
       +-----------------------+
       | Extract Complete Lines|
       +-----------+-----------+
                   |
                   v
       +-----------------------+
       | Dynamically Executed  |
       |  Transform Function   |
       +-----------+-----------+
                   |
                   v
       +-----------------------+
       | Output Sink (.jsonl)  |
       +-----------------------+

Full Script

#!/usr/bin/env python3
"""
Real-Time CSV Stream Processor
Tails a growing CSV file, dynamically applies pandas transformation logic,
and writes structured records to a JSON Lines or CSV output sink.
"""

import argparse
import importlib.util
import io
import os
import pathlib
import signal
import sys
import time
from typing import Callable, Optional
import pandas as pd

# Environment variable configurations with fallback defaults
DEFAULT_POLL_INTERVAL = float(os.environ.get("CSV_STREAM_POLL_INTERVAL", "1.0"))
DEFAULT_BATCH_SIZE = int(os.environ.get("CSV_STREAM_BATCH_SIZE", "1000"))

# Global running flag for handling signals gracefully
RUNNING = True


def handle_shutdown(signum: int, frame: object) -> None:
    """Handles SIGINT/SIGTERM signals to enable clean exit loops."""
    global RUNNING
    print("\n[INFO] Shutdown signal received. Finishing batch and exiting...", file=sys.stderr)
    RUNNING = False


def load_transform_function(transform_path: Optional[str]) -> Callable[[pd.DataFrame], pd.DataFrame]:
    """
    Dynamically loads a Python module and retrieves its 'transform' function.
    If no path is provided, returns an identity pass-through function.
    """
    if not transform_path:
        return lambda df: df

    path = pathlib.Path(transform_path).resolve()
    if not path.exists():
        raise FileNotFoundError(f"Transform script not found: {path}")

    module_name = path.stem
    spec = importlib.util.spec_from_file_location(module_name, path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Could not load spec from script: {path}")

    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    if not hasattr(module, "transform"):
        raise AttributeError(f"Module '{path}' must define a function named 'transform(df: pd.DataFrame) -> pd.DataFrame'")

    return getattr(module, "transform")


def write_to_sink(df: pd.DataFrame, output_path: pathlib.Path) -> None:
    """Appends transformed DataFrame records to the designated output sink."""
    if df.empty:
        return

    output_path.parent.mkdir(parents=True, exist_ok=True)
    
    if output_path.suffix.lower() == ".jsonl":
        df.to_json(output_path, orient="records", lines=True, mode="a")
    else:
        # For CSV, write header only if the destination file doesn't exist yet
        write_header = not output_path.exists()
        df.to_csv(output_path, mode="a", index=False, header=write_header)


def process_stream(
    input_path: pathlib.Path,
    output_path: pathlib.Path,
    transform_fn: Callable[[pd.DataFrame], pd.DataFrame],
    watch: bool,
    poll_interval: float,
    batch_size: int,
) -> None:
    """Main streaming engine loop tracking file byte offsets."""
    if not input_path.exists():
        raise FileNotFoundError(f"Input file not found: {input_path}")

    print(f"[INFO] Tailing stream from: {input_path}")
    print(f"[INFO] Streaming output to: {output_path}")

    header_line: Optional[str] = None
    offset = 0

    # Extract initial CSV header
    with open(input_path, "r", encoding="utf-8") as f:
        header_line = f.readline().strip()
        offset = f.tell()

    if not header_line:
        raise ValueError(f"Input file {input_path} is empty or missing headers.")

    while RUNNING:
        try:
            current_size = input_path.stat().st_size

            # Detect file truncation or rotation
            if current_size < offset:
                print("[WARN] File truncated or rotated. Resetting byte offset to start.", file=sys.stderr)
                offset = 0
                with open(input_path, "r", encoding="utf-8") as f:
                    header_line = f.readline().strip()
                    offset = f.tell()

            # Read new bytes if file has grown
            if current_size > offset:
                with open(input_path, "rb") as f:
                    f.seek(offset)
                    raw_data = f.read(current_size - offset)

                # Ensure we only process up to the last complete newline byte
                last_newline = raw_data.rfind(b"\n")
                if last_newline != -1:
                    bytes_to_process = raw_data[: last_newline + 1]
                    offset += last_newline + 1

                    text_chunk = bytes_to_process.decode("utf-8", errors="replace")
                    
                    # Prepend header to construct valid standalone CSV block
                    full_csv_buffer = f"{header_line}\n{text_chunk}"
                    
                    # Parse buffer with pandas
                    df = pd.read_csv(io.StringIO(full_csv_buffer))

                    if not df.empty:
                        # Process in chunk sizes if batching is specified
                        for i in range(0, len(df), batch_size):
                            batch = df.iloc[i : i + batch_size].copy()
                            transformed_df = transform_fn(batch)
                            write_to_sink(transformed_df, output_path)
                            print(f"[INFO] Processed {len(transformed_df)} records.")

        except Exception as err:
            print(f"[ERROR] Engine encountered error during stream cycle: {err}", file=sys.stderr)

        if not watch:
            break

        time.sleep(poll_interval)


def main() -> None:
    parser = argparse.ArgumentParser(description="Real-Time CSV Stream Processor")
    parser.add_argument("--input", required=True, type=pathlib.Path, help="Path to input source CSV file")
    parser.add_argument("--output", required=True, type=pathlib.Path, help="Path to output sink (.jsonl or .csv)")
    parser.add_argument("--transform", type=str, default=None, help="Path to python script with transform(df) function")
    parser.add_argument("--watch", action="store_true", help="Continuously watch/tail the input file for changes")
    parser.add_argument("--poll-interval", type=float, default=DEFAULT_POLL_INTERVAL, help="Poll interval in seconds")
    parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE, help="Maximum batch size per chunk")

    args = parser.parse_args()

    # Register OS signals for graceful shutdown
    signal.signal(signal.SIGINT, handle_shutdown)
    signal.signal(signal.SIGTERM, handle_shutdown)

    try:
        transform_fn = load_transform_function(args.transform)
        process_stream(
            input_path=args.input,
            output_path=args.output,
            transform_fn=transform_fn,
            watch=args.watch,
            poll_interval=args.poll_interval,
            batch_size=args.batch_size,
        )
    except Exception as e:
        print(f"[FATAL] Initialization or execution error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

Configuration

The script behavior can be configured either via environment variables or explicitly through command-line flags.

Usage Examples

1. Simple Single-Pass File Conversion (CSV to JSONL)

Extract records from a static CSV file and convert them directly into JSON Lines format:

python3 csv_stream.py --input system_events.csv --output sink.jsonl

Expected Output:

[INFO] Tailing stream from: system_events.csv
[INFO] Streaming output to: sink.jsonl
[INFO] Processed 150 records.

2. Continuous Real-Time Processing with Custom Transformation

Define custom enrichment logic inside a file named transform.py:

import pandas as pd

def transform(df: pd.DataFrame) -> pd.DataFrame:
    # Filter out low severity events and enrich data
    df = df[df["severity"] == "ERROR"].copy()
    df["processed_at"] = pd.Timestamp.utcnow()
    return df

Execute the stream watcher using the dynamic script:

python3 csv_stream.py \
  --input /var/log/app_events.csv \
  --watch \
  --transform transform.py \
  --output /var/log/alerts.jsonl \
  --poll-interval 0.5

Expected Output:

[INFO] Tailing stream from: /var/log/app_events.csv
[INFO] Streaming output to: /var/log/alerts.jsonl
[INFO] Processed 4 records.
[INFO] Processed 12 records.
^C
[INFO] Shutdown signal received. Finishing batch and exiting...

Error Handling

• Incomplete Row Writes: Appending to files in real-time can result in reading partial CSV lines. The script isolates complete lines by slicing buffer contents strictly up to the final byte newline (b"\n"), maintaining offset integrity for future cycles.
• File Truncation / Rotation: If a log rotator truncates or recreates the target file (current_size < offset), the script automatically detects the anomaly, resets the byte offset to zero, re-reads the header, and resumes processing without crashing.
• Malformed Custom Transforms: If --transform references a script lacking a transform(df) function signature or containing syntax errors, execution fails immediately with a descriptive error.
• Signal Interruption: Interruption signals (SIGINT / SIGTERM) trip a global flag RUNNING = False. The loop completes the active record batch cleanly before exiting, avoiding output sink corruption.

Extending Ideas

1. Persistent Offset Storage: Save the byte offset value to an On-Disk SQLite database or state file (.offset). This allows the script to safely resume streaming after process restarts or daemon crashes.
2. Schema Validation via Pydantic: Integrate pydantic within the data pipeline loop to validate column data types and drop invalid schema rows before output writing.
3. Kafka/RabbitMQ Streaming Sink: Replace the file writing sink (write_to_sink) with an event queue publisher (e.g., confluent-kafka or pika) to push transformed frames directly to a message broker.
4. Prometheus Metrics Exporter: Bind prometheus_client counters to expose metrics for records read, batch processing latencies, and error rates over an HTTP endpoint.