1.2 Ingest, Store, and Retain: Building a Replayable Data Pipeline

A contract defines what a record represents. Operations expose the rest of the lifecycle: queues retry, batch jobs can partially fail, and historical partitions need replay. A data lifecycle is therefore a set of workflows that must be replayable, observable, and deletable, not a list of seven static nouns.

Learning Objectives

• Choose between batch and stream processing based on latency, throughput, and consistency requirements;
• Design clear boundaries between raw, validated, and serving layers;
• Handle failures using checkpoints, idempotency, and reconciliation mechanisms;
• Establish verifiable workflows for retention, archiving, and deletion.

1. Batch and Streaming Are Choices in Data Boundary

Batch processing handles bounded datasets: a single file on a given day, a snapshot, or a defined date range. Streaming processing deals with unbounded, continuously arriving events, where systems establish temporary computable boundaries using windows and watermarks.

Dimension	Batch	Streaming
Result latency	Minutes to days	Milliseconds to minutes
Input boundary	Clearly defined end	Continuously arriving
Failure recovery	Redo the batch	Checkpoint + replay
Late data	Backfill in the next batch	Watermark / allowed lateness
Operational complexity	Generally lower	More complex due to state, ordering, and backpressure

A single system can "ingest in streaming mode and analyze in batch," or use micro-batching to achieve near real-time performance. Business requirements (like triggering alerts within five minutes) do not necessitate converting all historical analysis to streaming processing.

The latency ranges in the table are typical workloads, not definitions or guarantees. Streaming without event-time aggregation may not use windows at all. A watermark estimates progress; it cannot prove that no older event will ever arrive.

2. Establish Batch Identity and Inventory

When reading daily CSV files, at a minimum record:

batch_id
source_uri
source_version / etag
expected_size
checksum
observed_rows
schema_version
started_at / completed_at
status

from io import BytesIO
import hashlib
import pandas as pd

# A small immutable CSV snapshot, included so the example runs as written.
payload = b"mission_id,status\nM-1042,completed\nM-1043,failed\n"
digest = hashlib.sha256(payload).hexdigest()
frame = pd.read_csv(BytesIO(payload), dtype="string", keep_default_na=False)

manifest = {
    "source": "in-memory-example",
    "sha256": digest,
    "bytes": len(payload),
    "rows": len(frame),
    "columns": list(frame.columns),
}
assert manifest["rows"] == 2
print(manifest)

This small pandas example hashes and parses the same bytes. For a local file, read a fixed snapshot into payload; independently hashing and reopening a mutable path could describe two different versions. Large inputs need chunked hashing and parsing over an immutable object/version, not just two streaming reads of a changing file. DataFrame.info() prints a report and returns None.

3. Reconcile the End-to-End Data

TCP or message middleware may handle retry mechanisms at the transport layer, but business data can still be lost or duplicated due to production transactions, serialization, filtering, consumer offset-commit timing, and failures during target write operations.

Verification should include:

• Source record count versus target accepted or rejected counts;
• Event ID count before and after deduplication;
• Minimum and maximum timestamps per partition;
• Conserved aggregates such as amounts and counts;
• Dead-letter queue size and root causes;
• Source offset versus checkpoint position.

Simple equality of line counts is not sufficient: one lost record and one duplicated record can still result in the same total. Stable identifiers, checksums, and business-level aggregations must be used together to ensure validity.

For example, source IDs [a, b, c] and target IDs [a, b, b] both contain three rows. A distinct-ID check exposes the duplicate; a source-to-target anti-join identifies missing c. A matching amount sum is also insufficient if duplicated and missing amounts happen to match.

4. Layered Storage and Responsibility Isolation

A common (but by no means exclusive) layering approach:

raw        retains the original received representation and provenance
validated  passes schema and foundational quality checks, isolating invalid records
curated    achieves semantic consistency, deduplication, and clear business rule application
serving    organizes data for reporting, APIs, or model consumption

Each layer must have well-defined inputs, outputs, and reconstruction mechanisms. Don’t treat "bronze/silver/gold" as immutable truths; the names themselves are irrelevant. What matters is the reproducible boundary and the quality commitments at each stage.

Storage selection is determined jointly by access patterns, update semantics, scale, latency, and cost. Columnar formats are well-suited for scanning and analytical workloads, but excessive small files can degrade metadata performance and scheduling efficiency. Row-based databases excel at point queries and transactions, yet this does not preclude them from supporting aggregation operations. Avoid concluding that a change in storage is necessary simply because data is aggregated daily.

5. Processing Steps Must Be Idempotent

If a task fails mid-execution, retrying should not accumulate or duplicate results. Common strategies include:

• Write to a temporary location per partition, then atomically publish upon success;
• Perform an upsert operation keyed by business identifier;
• Ensure output paths include deterministic run and version identifiers;
• Coordinate checkpoints with external side effects;
• Record the lineage from input version to output version.

While "delete the target partition first, then rewrite" is simple, it introduces a temporary gap and risks losing previous results if rewriting fails after deletion. A safer approach is to write a new version, validate its integrity, and then atomically update the metadata pointer.

An upsert is idempotent only if repeating it leaves the same result: total = total + delta still accumulates twice. An atomic publication pointer also requires readers to honor it. Object-store rename may be copy-and-delete, and concurrent backfills need version checks so an older run cannot publish over a newer result.

6. Backfill and Daily Operations Use the Same Logic

When fixing historical bugs, backfilling is required. If backfilling is implemented through ad-hoc scripts, rules quickly drift. Instead, pipelines should be parameterized:

process(start_time, end_time, input_version, code_version)

Before performing a backfill, evaluate the following:

• Which partitions will be rewritten;
• Whether downstream components will automatically trigger;
• Whether current data is being mixed with results from old and new rules;
• Whether the backfill will take resources away from online tasks;
• How to compare results from the old version versus the new version;
• How to roll back the published version pointer.

7. Observability Focuses on Data State

Beyond monitoring CPU, memory, and task success rates, you must also track:

• freshness: How long ago was the most recent complete data snapshot?
• volume: The number of records and bytes compared to historical baselines;
• validity: Schema violations, out-of-range values, and enum violations;
• completeness: Whether critical fields and partitions are present;
• uniqueness: Whether business keys are duplicated;
• distribution: Whether data distribution has drifted abnormally;
• reconciliation: Whether the source and target datasets remain balanced.

Thresholds should account for weekdays, seasonality, and business releases. Alerting simply on “10% less than yesterday” will generate noise during weekends when data activity is naturally lower.

8. Retention and Archiving Driven by Obligation

Retention periods should be derived from a balance of:

• Product and analytical requirements;
• Legal preservation mandates and deletion rights;
• Debugging and auditing windows;
• Historical data needed for replay;
• Storage and recovery costs.

Before archiving, verify readability: the existence of an object does not guarantee that years later, its schema, keys, and software stack will still be interpretable. Conduct regular recovery drills and preserve format, schema, and integrity metadata.

Compression ratios depend on data distribution and encoding schemes, no universal promise of "80% reduction" can be made. Perform sample testing with gzip and zstd compression and Parquet encoding options, evaluating trade-offs between cost and access patterns.

9. Removing Objects Requires Covering Derivation and Backup

Object removal includes the following chain:

Original object → Cleaning table → Aggregation table → Features → Search index → Cache → Backup

Deleting a file name does not equate to immediate physical unrecoverability. In modern SSDs, object storage, and managed databases, erasure is determined by the underlying implementation, repeated overwriting is not guaranteed to be reliable or effective. Common approaches include:

• Logically deleting an object and then having the storage lifecycle management remove its version;
• Crypto erasure: destroying the independent encryption keys;
• Using the vendor-provided media sanitization and erasure procedures;
• Setting immutable but time-limited expiration windows for backups;
• Retaining audit logs of deletions that do not contain sensitive information.

Legally, what must be deleted and when it must be completed depends on applicable regulations and roles, and must be confirmed by compliance and security teams. No tutorial can provide a universal deadline.

Cryptographic erasure depends on the encryption and key lifecycle: all relevant key copies and plaintext copies must be accounted for. Restoring an older backup should replay the deletion ledger before serving reads. NIST SP 800-88 Rev. 2 describes media sanitization as a validated process, not just a file-removal command.

Common Misconceptions

• Task success implies data integrity: A successful status does not guarantee semantic validation.
• Equal row counts mean no data loss: Duplicates can cancel out missing entries.
• The original layer is always the safest: It can actually expand exposure to sensitive data and compliance risks.
• rm or repeated overwrites work for all storage: Behavior differs across media, versioning, and hosted layers.

Exercise

1. Design a batch manifest and idempotent retry strategy for daily snapshots.
2. Construct a counterexample where rows are equal in count but differ by a single missing or extra entry.
3. Outline the steps for publishing, verifying, and rolling back a historical backfill.
4. Diagram all derived data nodes that must be covered when a user submits a deletion request.

Summary

An operable data pipeline must clearly define input versions, processing state, output versions, and failure recovery mechanisms. The choice between batch and stream processing determines the boundary of the pipeline. Idempotency and replay handling are essential for fault tolerance, while quality monitoring ensures the results are trustworthy. Retention and deletion policies govern when data exits the system.

The next chapter covers data cleaning: identify missing-data mechanisms, type ambiguities, and the assumptions made by each repair rule before deciding whether median imputation is appropriate.