NDJSON for Big Data and Streaming Pipelines

Use record framing to bound the working set

A single JSON array can contain millions of records, but a conventional whole-document parser builds the entire value before the application can process it. Newline-delimited JSON instead places one complete JSON value on each physical line. The consumer can process records incrementally when it combines that framing with bounded reads and downstream backpressure.

The newline is a framing delimiter outside each record. A newline inside a string must be escaped, so pretty-printing one record across several physical lines breaks the format. A stream of objects separated by commas is not NDJSON. Define whether your application accepts any JSON root value or requires an object, because the framing convention and the application schema are separate concerns.

Streaming is not automatic merely because the file extension changes. Reading the entire file and splitting it into lines still retains the full input, and collecting every parsed record into a list defeats incremental processing. Design the whole pipeline around bounded live state. The Python streaming guide contrasts record-oriented input with incremental parsing of a large array when you cannot change the producer.

Define byte, line, and encoding policies

Use UTF-8 and establish a clear policy for blank lines, final newlines, and malformed records. JSON Lines treats every line as a valid value, so blank lines are not valid records. A trailing newline is useful for concatenation and append workflows. Some import systems tolerate a missing final newline, but that compatibility should not conceal a partially written final record.

Limit record bytes before parsing. A line-oriented API that reads until a newline can allocate an arbitrarily large string if a producer never sends one. Use a bounded line read or incremental buffer with a maximum size. Also bound total file size or job duration when the workflow requires it. A small record limit does not constrain an endless stream.

Decode strictly and preserve chunk boundaries correctly. Network chunks are not records and can split a multi-byte character or a newline sequence. Buffer bytes until a complete bounded record is available, or use an incremental decoder with explicit framing. The Unicode guide explains why replacement decoding can silently alter identifiers before JSON parsing even begins.

A bounded executable NDJSON reader

The following Python program reads one record at a time with a byte cap, rejects blank records and non-object roots, and demonstrates consumption without retaining the entire stream. Save it as ndjson_reader.py and run it directly. It uses an in-memory byte stream for a self-contained test; the same reader accepts a binary file opened from disk.

The maximum is defined over the physical record including its newline when present. That keeps the boundary simple and testable. The reader rejects a line as soon as the bounded read shows that it exceeds the cap. It does not continue silently after malformed JSON because this example chooses fail-fast ingestion. A quarantine policy would be a different documented mode with its own checkpoint semantics.

Do not print or retain full records when reporting failures in a sensitive pipeline. A line number, source identifier, and bounded error category are often enough to locate the producer problem. If a dead-letter store is required, protect it like the original data and set a retention policy. Operational convenience should not turn rejected records into an unbounded secondary dataset.

import io
import json

def records(stream, max_record_bytes=1_048_576):
    line_number = 0
    while True:
        raw = stream.readline(max_record_bytes + 1)
        if not raw:
            return
        line_number += 1
        if len(raw) > max_record_bytes:
            raise ValueError(f"record {line_number}: too large")
        if not raw.strip():
            raise ValueError(f"record {line_number}: blank")
        value = json.loads(raw.decode("utf-8"))
        if not isinstance(value, dict):
            raise ValueError(f"record {line_number}: object required")
        yield line_number, value

source = io.BytesIO(b'{"id":"a","amount":2}\n{"id":"b","amount":3}\n')
total = 0
for number, value in records(source):
    total += value["amount"]
assert total == 5
print("total =", total)

try:
    list(records(io.BytesIO(b'{"long":"abcdef"}\n'), 8))
except ValueError:
    print("oversized record rejected")

Backpressure must reach the source

A parser can yield one record at a time while an unbounded task queue still accumulates every record downstream. Limit the number of active writes and pending batches. When the sink slows, the reader should stop advancing rather than continue allocating work. This is the practical meaning of backpressure across the pipeline.

Choose batches by both record count and bytes. A thousand tiny events and a thousand large documents have very different memory and network costs. Flush on a time bound when low traffic would otherwise leave a small batch waiting indefinitely. Keep retry batches bounded as well; a failing sink should not create a second unbounded queue.

Measure source read rate, sink completion rate, queue depth, and oldest pending-record age. Throughput alone can look healthy while latency grows inside a queue. For tenant-shared ingestion, prevent one large import from monopolizing all workers. The resource-control guide covers admission limits that complement streaming. A memory-efficient parser is only one component of a memory-efficient service.

Append-only logs need durability and recovery rules

Appending a complete JSON record and newline is convenient, but a process can fail during a write. Recovery must distinguish a complete final record from a truncated tail. Do not simply skip every malformed line, because corruption in the middle of a file may indicate a serious producer or storage problem. Define whether the file is a best-effort log or a durable business record.

Single-writer discipline simplifies record boundaries. Multiple processes appending without coordination can interleave data depending on the platform and write pattern. Use a logging system or explicit synchronization rather than assuming all appends are atomic. Durability also depends on flushing and storage behavior; a successful language-level write does not always mean the data has reached durable media.

Checkpoint only after the corresponding sink effect is durable under your chosen semantics. A byte offset can identify a position in an immutable file, but compressed streams and rotated files need additional identity information. Record the source version, partition, or file identity alongside the position. On restart, verify that the checkpoint refers to the same data rather than resuming blindly into a replaced file.

Kafka messages are not NDJSON lines by necessity

Kafka already frames records, so putting one JSON document in each message does not require a newline delimiter. NDJSON is useful for file exports, import tooling, or batches inside a message, but the broker's record boundary is the primary transport frame. Avoid adding an extra framing layer without a concrete reason.

Choose keys according to ordering and partition requirements. Ordering is not global merely because every payload is JSON. A consumer processing several partitions concurrently must preserve any required per-key or per-partition sequencing. Include an event identifier and schema version where the application needs deduplication and evolution, but do not assume those fields enforce anything by themselves.

Offset commits and sink writes need a failure model. Committing before a durable sink effect risks loss; committing afterward can lead to replay if the process fails between the two steps. Design idempotent sink behavior or a transactional strategy appropriate to the system. Label the actual guarantee rather than calling every streaming pipeline exactly once. The serialized format does not solve distributed commit coordination.

Elasticsearch bulk framing has its own contract

The bulk API uses action metadata lines and, for applicable actions, source lines. It is not simply an arbitrary NDJSON file of documents posted unchanged. Index and create operations pair an action with a source, while delete has no source line. The request must follow the API's framing rules, including its final newline requirement.

Inspect item-level results even when the HTTP request succeeds. A batch can contain successful and failed operations. Retry only the operations appropriate for retry, preserving identifiers and semantics. Retrying the whole batch indiscriminately can duplicate effects or waste capacity. Distinguish transient failures from mapping errors that will not improve with repetition.

Use bounded batches, stable document identifiers when idempotence is intended, and explicit mapping expectations. Dynamic field explosion can become a storage and query problem even if every record is small and valid JSON. Monitor rejected items and dead-letter volume, and stop or throttle an import whose failure rate indicates a bad schema. A fast parser should not help a broken producer overwhelm the destination faster.

Test restart behavior, not only steady-state speed

A useful ingestion test stops the process after reading a record, after writing to the sink, and before checkpointing. Restart and inspect whether the system loses or duplicates effects under the documented policy. Use deterministic synthetic identifiers so replay is observable. Ordinary throughput tests rarely expose these boundary failures.

Include malformed UTF-8, blank lines, oversized records, a truncated final record, and a slow sink. Verify that memory remains bounded and that errors identify the source position without leaking contents. Test compressed input separately because decompression expansion changes the resource budget and checkpoint strategy.

Retain the parser settings, schema version, batch policy, and sink semantics in the job configuration. A rerun should not silently use different validation rules from the original import. For long-lived pipelines, make schema transitions explicit and preserve enough metadata to replay historical records intentionally. NDJSON provides useful framing, but dependable ingestion comes from bounded work, clear durability boundaries, and tested recovery behavior.

Compression and checkpoints need compatible framing

A byte offset in an uncompressed immutable file can identify a record boundary, but an offset in a compressed stream is not generally enough to restart decoding at that location. If resumability matters, use independently compressed chunks, a seekable format with appropriate indexing, or a staging process that records valid restart points.

Keep source identity with the checkpoint. A path can be reused after rotation or replacement, so filename alone is weak evidence that the bytes are unchanged. Use a version identifier, immutable object key, or verified metadata appropriate to the storage system. Reject a resume attempt when the source no longer matches the checkpoint contract.

Distinguish a clean end of stream from an interrupted transfer. An importer that accepts a missing final newline may still need to know whether the transport completed successfully. A valid last record does not prove that all expected records arrived. Use expected counts, manifests, or transport integrity where the workflow requires completeness.

Apply decompressed-byte and job-duration budgets even when individual lines are bounded. An endless sequence of small valid records can still consume unlimited time or storage. Streaming makes memory manageable; it does not make the total operation free.

Make dead-letter handling a bounded operational workflow

A dead-letter destination needs a schema, retention policy, and owner. Store the source identity, record position, failure category, and processing version alongside any protected original data. Without those fields, replay becomes guesswork and operators may be unable to determine whether a record was already applied.

Separate transient sink failures from permanent record failures. A temporary connection problem should not label valid data as permanently invalid, while a mapping error should not be retried forever. Use bounded retries with a clear transition to operator review or job failure. Keep retry queues from growing faster than the main pipeline.

When replaying corrected records, preserve the original business identifier and apply the same idempotency policy as normal ingestion. A replay tool should not accidentally bypass validation because it is considered internal. Record the correction or schema version so future investigators can understand why the second attempt succeeded.

Test a batch containing both valid and invalid records. Verify that item-level success is tracked correctly and that the checkpoint does not skip failed work unintentionally. For systems that stop on the first error, ensure previously committed effects can be replayed safely after the producer fixes the input.

Finally, set an alert on failure proportion as well as absolute count. A small import with every record failing can be more urgent than a large import with a few isolated failures. Stop systematic bad data early rather than filling a dead-letter store efficiently.

Engineering Comparison

Framing, durability, and exactly-once effects are separate properties. The runnable reader demonstrates bounded records, not a distributed transaction protocol.