Python Script for Distributed Log Processing in Big Data

Why distributed log processing needs a script first

When log volume outgrows a single host, the real problem is not just “reading files faster.” It is making sure records are collected, partitioned, validated, processed, and written back without losing events, duplicating work, or overwhelming storage and downstream systems. In distributed environments, a small scripting mistake can turn into corrupted outputs, noisy retries, or expensive reprocessing.

This guide shows how to build a practical Python script skeleton for distributed log processing in a big data workflow. After reading it, you will be able to decide whether this approach fits your pipeline, run a safe baseline script, validate input and output behavior, and identify what must change before production use.

What this Python script does

The script below is designed for a common pattern: take one or more log sources, parse each line as structured data, filter or transform records, and write partitioned output that can be consumed by downstream analytics jobs.

It does:

• Validate input paths and required parameters before processing starts.
• Read plain text or gzipped log files line by line.
• Parse JSON log lines safely, while preserving malformed lines for review.
• Support dry-run mode by default so no output is written unless explicitly enabled.
• Emit structured summary output for automation.
• Use conservative batch writes and clear error handling.

It does not:

• Provide cluster orchestration, job scheduling, or autoscaling.
• Replace a log shipping platform, message queue, or distributed execution framework.
• Guarantee exactly-once delivery across failures.
• Infer schemas beyond the fields you define.

If your environment already has a production big data control plane, use this script as the processing core inside a job runner, container task, or batch execution wrapper. Before production use, verify architecture, scaling, monitoring, and rollback readiness with a Big Data Production Readiness Checklist.

Assumptions, requirements, and permissions

This script assumes the following:

• Input logs are available on local disk, shared storage, or mounted object storage.
• Log lines are newline-delimited.
• Structured records are JSON objects, or at minimum contain a timestamp and message field.
• The processing node can read source paths and, when not in dry-run mode, write to the destination path.

Requirements:

• Python 3.10 or later.
• Standard library only for the base version shown here.
• Sufficient disk space for temporary output files if you use atomic writes.
• Read permission on all input files.
• Write permission on the output directory when --write is enabled.

If your log format depends on vendor-specific exports or agent-specific fields, confirm field names and timestamp format before relying on the parser.

The Python script skeleton

#!/usr/bin/env python3
"""Distributed log processing skeleton for batch log normalization.

Safe defaults:
- dry-run mode by default
- input validation before processing
- atomic output writes when enabled
- malformed records preserved in a separate error report
"""

from __future__ import annotations

import argparse
import gzip
import hashlib
import json
import os
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, Iterator


@dataclass
class LogRecord:
    source_file: str
    line_number: int
    timestamp: str | None
    level: str | None
    message: str | None
    raw: str
    parse_error: str | None = None


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Process log files into normalized JSONL output."
    )
    parser.add_argument(
        "--input",
        nargs="+",
        required=True,
        help="One or more input files or directories containing log files",
    )
    parser.add_argument(
        "--output-dir",
        required=True,
        help="Directory where normalized output will be written",
    )
    parser.add_argument(
        "--write",
        action="store_true",
        help="Write output files. If omitted, run in dry-run mode.",
    )
    parser.add_argument(
        "--pattern",
        default="*.log",
        help="Glob pattern used when input is a directory (default: *.log)",
    )
    parser.add_argument(
        "--max-lines-per-file",
        type=int,
        default=500000,
        help="Rotate output files after this many lines",
    )
    return parser.parse_args()


def validate_args(args: argparse.Namespace) -> None:
    if not args.input:
        raise ValueError("At least one input path is required")
    if args.max_lines_per_file <= 0:
        raise ValueError("--max-lines-per-file must be greater than zero")

    for item in args.input:
        path = Path(item)
        if not path.exists():
            raise FileNotFoundError(f"Input path does not exist: {path}")

    out_dir = Path(args.output_dir)
    if args.write:
        out_dir.mkdir(parents=True, exist_ok=True)
        if not os.access(out_dir, os.W_OK):
            raise PermissionError(f"Output directory is not writable: {out_dir}")


def iter_input_files(inputs: Iterable[str], pattern: str) -> Iterator[Path]:
    for item in inputs:
        path = Path(item)
        if path.is_dir():
            yield from sorted(path.glob(pattern))
        else:
            yield path


def open_text(path: Path):
    if path.suffix == ".gz":
        return gzip.open(path, "rt", encoding="utf-8", errors="replace")
    return path.open("r", encoding="utf-8", errors="replace")


def normalize_record(line: str, source_file: str, line_number: int) -> LogRecord:
    raw = line.rstrip("\n")
    try:
        obj = json.loads(raw)
        if not isinstance(obj, dict):
            raise ValueError("JSON root must be an object")
        return LogRecord(
            source_file=source_file,
            line_number=line_number,
            timestamp=obj.get("timestamp"),
            level=obj.get("level"),
            message=obj.get("message"),
            raw=raw,
        )
    except Exception as exc:
        return LogRecord(
            source_file=source_file,
            line_number=line_number,
            timestamp=None,
            level=None,
            message=None,
            raw=raw,
            parse_error=str(exc),
        )


def output_name(source_file: Path, shard: int) -> str:
    digest = hashlib.sha1(str(source_file).encode("utf-8")).hexdigest()[:10]
    return f"normalized-{source_file.stem}-{digest}-part{shard:04d}.jsonl"


def write_jsonl_atomic(target: Path, records: list[LogRecord]) -> None:
    tmp = target.with_suffix(target.suffix + ".tmp")
    with tmp.open("w", encoding="utf-8") as f:
        for record in records:
            f.write(json.dumps(asdict(record), ensure_ascii=False) + "\n")
    tmp.replace(target)


def process_file(input_path: Path, output_dir: Path, write: bool, max_lines_per_file: int) -> dict:
    total = 0
    malformed = 0
    shard = 1
    buffer: list[LogRecord] = []

    with open_text(input_path) as f:
        for line_number, line in enumerate(f, start=1):
            total += 1
            record = normalize_record(line, str(input_path), line_number)
            if record.parse_error:
                malformed += 1
            buffer.append(record)

            if write and len(buffer) >= max_lines_per_file:
                target = output_dir / output_name(input_path, shard)
                write_jsonl_atomic(target, buffer)
                buffer.clear()
                shard += 1

    if write and buffer:
        target = output_dir / output_name(input_path, shard)
        write_jsonl_atomic(target, buffer)

    return {
        "input_file": str(input_path),
        "lines_processed": total,
        "malformed_lines": malformed,
        "written": write,
        "status": "ok",
    }


def main() -> int:
    args = parse_args()
    try:
        validate_args(args)
        output_dir = Path(args.output_dir)
        results = []

        for input_file in iter_input_files(args.input, args.pattern):
            if not input_file.is_file():
                continue
            result = process_file(input_file, output_dir, args.write, args.max_lines_per_file)
            results.append(result)
            print(json.dumps(result, sort_keys=True))

        summary = {
            "processed_at": datetime.now(timezone.utc).isoformat(),
            "files": len(results),
            "total_lines": sum(r["lines_processed"] for r in results),
            "total_malformed": sum(r["malformed_lines"] for r in results),
            "mode": "write" if args.write else "dry-run",
        }
        print(json.dumps(summary, sort_keys=True))
        return 0

    except (ValueError, FileNotFoundError, PermissionError) as exc:
        print(json.dumps({"status": "error", "error": str(exc)}), sort_keys=True)
        return 2
    except Exception as exc:
        print(json.dumps({"status": "error", "error": f"unexpected: {exc}"}), sort_keys=True)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

How the script works in a distributed workflow

The script is intentionally single-process and file-oriented so it stays predictable. In a distributed environment, you run it multiple times against different partitions of data rather than letting it coordinate the cluster itself. That keeps the execution model simple and makes failures easier to isolate.

A common deployment pattern is:

1. Split logs by host, date, tenant, service, or shard.
2. Run this script per partition in parallel.
3. Write normalized JSONL output to a partitioned destination directory.
4. Aggregate or index the output with downstream batch or search jobs.

This model is safer than one large shared write stream because each job has a bounded input set and can be retried independently.

Expected input and output

Input is newline-delimited text files or .gz files. The default parser expects each line to be a JSON object similar to this:

{"timestamp":"2026-07-05T10:00:00Z","level":"INFO","message":"service started","host":"app-01"}

If a line is not valid JSON, the script does not stop. It preserves the raw line and tags the record with parse_error so you can review bad data separately.

Output is JSON Lines, one normalized object per line, with metadata fields added:

{"line_number":1,"level":"INFO","message":"service started","parse_error":null,"raw":"{\"timestamp\":\"2026-07-05T10:00:00Z\",\"level\":\"INFO\",\"message\":\"service started\",\"host\":\"app-01\"}","source_file":"/data/logs/app-01.log","timestamp":"2026-07-05T10:00:00Z"}

This output format is easy to reprocess, compress, index, or load into analytics storage.

Example command and expected output

Dry-run mode is the default, so the script validates and reports without writing files:

python3 log_processor.py --input /data/logs/app-01.log /data/logs/app-02.log --output-dir /tmp/normalized

Example output:

{"input_file": "/data/logs/app-01.log", "lines_processed": 1200, "malformed_lines": 3, "status": "ok", "written": false}
{"input_file": "/data/logs/app-02.log", "lines_processed": 980, "malformed_lines": 1, "status": "ok", "written": false}
{"files": 2, "mode": "dry-run", "processed_at": "2026-07-05T10:05:00+00:00", "total_lines": 2180, "total_malformed": 4}

To enable writing:

python3 log_processor.py --input /data/logs/ --output-dir /data/normalized --write --pattern '*.log'

What to change before production use

The skeleton is safe, but production use typically requires more than parsing and writing files. Review these items before relying on it in a live pipeline:

• Replace local file iteration with your actual storage layer if input lives in object storage, distributed filesystem, or a streaming sink.
• Decide whether malformed records should be preserved, quarantined, or dropped.
• Add schema validation for required fields such as timestamp, service, host, or request_id.
• Define partitioning rules for output paths by date, tenant, region, or severity.
• Add retries and idempotency if the script is wrapped by an external scheduler.
• Add checksum or manifest generation if downstream consumers require completeness checks.
• Add metrics export if you need centralized monitoring for throughput, failures, or malformed rate.

If your processing flow touches security-sensitive data, consider masking or redacting fields before writing output. For parsing and validation patterns that can be adapted to other automation tasks, see JavaScript Script to Parse and Validate JSON Payloads when you need a comparable validation-first approach in another runtime.

Validation and testing steps

Before production, verify behavior with known-good and known-bad samples. A practical test set should include:

• A small valid JSONL file.
• A file with blank lines.
• A file with malformed JSON.
• A gzipped log file.
• An unreadable file to confirm permission handling.
• An output directory without write permission when testing failure behavior.

Suggested checks:

python3 -m py_compile log_processor.py
python3 log_processor.py --input samples/good.log --output-dir /tmp/normalized
python3 log_processor.py --input samples/bad.log --output-dir /tmp/normalized
python3 log_processor.py --input samples/good.log --output-dir /tmp/normalized --write

Validation criteria:

• Dry-run mode prints summaries and exits with code 0 on valid input.
• Invalid paths fail before any processing starts.
• Malformed lines are counted and preserved in output.
• Write mode creates JSONL files only when explicitly enabled.
• Output filenames remain deterministic for the same input path and shard number.

Security and privacy notes

Log processing often exposes tokens, emails, internal hostnames, IP addresses, user IDs, and request headers. The safest approach is to minimize what you store and who can read it.

Key controls:

• Restrict read access to source logs and write access to normalized output.
• Avoid logging secrets in exception traces or debug output.
• Redact known sensitive fields before writing if your logs include authentication data.
• Keep temporary files on encrypted storage where required by policy.
• Ensure the script does not silently skip unexpected parse failures.

If the script will run in a regulated environment, validate retention rules and access controls before enabling write mode in production.

Cleanup and rollback guidance

Because the default mode is dry-run, rollback is mostly about stopping execution and removing partial outputs. If write mode is enabled, keep the output path partitioned so you can delete only the affected batch.

Recommended rollback steps:

1. Stop the job or disable the scheduler entry.
2. Remove incomplete output files for the affected partition.
3. Re-run the script in dry-run mode against the same input to confirm the issue is fixed.
4. Restore from a known-good backup or reprocess from the original source if needed.

Atomic file replacement in the script reduces the chance of partially written outputs, but it does not protect you from writing the wrong data. Partitioned output paths and deterministic file naming make cleanup much safer.

Common mistakes

Mistake / Why it matters / Better approach

• Assuming every line is valid JSON / One malformed record can break naive parsers or cause silent data loss / Keep parse errors in output and count them explicitly.
• Writing directly to final files without atomic replacement / Interrupted runs can leave partial files behind / Write to a temporary file first, then rename atomically.
• Enabling write mode by default / Accidental execution can overwrite or fill storage / Default to dry-run and require an explicit --write flag.
• Ignoring output permissions until runtime / Jobs fail after expensive processing starts / Validate read and write permissions before opening files.
• Using one giant output file for all partitions / Large reruns become slow and fragile / Partition output by input source or batch shard.
• Skipping malformed-line reporting / You lose evidence of data quality issues / Report malformed counts and preserve raw content for review.

Final takeaway

A distributed log processing Python script should be boring in the best way: validate early, process conservatively, preserve bad records for inspection, and keep write behavior explicit and atomic. If you can run it safely in dry-run mode, prove its output on representative samples, and confirm permissions and rollback boundaries before production, you have a script that is practical enough to operationalize and disciplined enough to trust.

Use this guidance together with async errors with Promises and git branch cleanup script to connect the workflow with related operational context already available on the site.