Your personal operating system for career transition.

Implement a JSONL Ingestion Pipeline with Error Isolation

JSONL (JSON Lines) is the standard format for AI datasets, evaluation sets, and training data. An ingestion pipeline reads JSONL records, validates them, transforms them, and writes valid records to an output file — while isolating failures per record so a single malformed entry does not abort the whole run.

What to build

Implement an IngestionPipeline class that:

1. __init__(schema: type[BaseModel], transform: Callable[[BaseModel], dict]) — accept a Pydantic model class for input validation and a transform function that converts a validated record to an output dict.

2. run(input_path: str, output_path: str, overwrite: bool = False) -> IngestionResult — read the JSONL file at input_path, validate each line against schema, apply transform, write valid records to output_path, and return an IngestionResult.

3. IngestionResult — a dataclass with:

   • total_lines: int — total lines read (including blank lines)
   • processed: int — successfully validated and transformed
   • failed: int — lines that failed parsing or validation
   • skipped: int — blank or whitespace-only lines
   • failures: list[dict] — each with line_number, raw, and error

4. Idempotency — if output_path already exists and overwrite=False, raise FileExistsError with a clear message. If overwrite=True, overwrite it.

5. Atomic write — write to a temp file first, then rename to output_path. This prevents partial output files from being left on disk if the process is interrupted.

Test schema and transform

class RawArticle(BaseModel):
    id: str = Field(min_length=1)
    title: str = Field(min_length=1)
    body: str = Field(min_length=50)
    source: str

def transform_article(article: RawArticle) -> dict:
    return {
        "id": article.id,
        "title": article.title.strip(),
        "body_length": len(article.body),
        "source": article.source,
    }

Why this matters

The difference between an idempotent pipeline and one that cannot be safely rerun is a detail that matters enormously at 3 AM when you need to reprocess a dataset. The atomic write prevents the worst case: a half-written output file that looks valid but contains garbage after the interrupt point.

Constraints

• Use pathlib.Path for file operations, not bare open().
• The temp file should be in the same directory as output_path to ensure the rename is atomic on the same filesystem.
• Empty lines and whitespace-only lines count as skipped, not failed.