Skip to main content

JSON format

The JSON deserializer (etl-json, feature json on the etl facade) decodes JSON payloads carried by a source into records — either your own serde types or dynamically-typed values. Three framings map onto the framework's "one payload → 0..N records" contract, and serde_json (stable 1.x) does the parsing.

Construct it from the pipeline's opaque section:

let deserializer = JsonDeserializerBuilder::from_component(deser_section)?
    .with_metrics(ctx.pipeline, "main")
    .build_serde::<Order>();

Configuration​

The deserializer: { json: ... } section deserializes into JsonSettings (crates/etl-json/src/config.rs); unknown fields are rejected.

deserializer:
  json:
    framing: single              # single | ndjson | array
    on_error: skip               # skip | fail
    reject_duplicate_keys: false # error on any repeated object key
KeyTypeDefaultMeaning
framingstringsingleHow a payload is split into documents: single, ndjson, or array.
on_errorstringskipPer-record policy for a document that does not parse or match the target type: skip (drop + count) or fail (surface a decode error).
reject_duplicate_keysboolfalseReject (rather than silently keep the last value of) any JSON object with a duplicate key, at any depth. Parses each document a second time for the check.

The three framings​

  • single — the whole payload is one JSON document → one record. The Kafka-message default. An empty or whitespace-only payload is a tombstone: zero records, no error.
  • ndjson — newline-delimited JSON (JSON Lines): one JSON value per \n-separated line → one record per line. Blank lines are skipped. This is the framing with per-line error isolation (a malformed line is skipped or fails on its own; the rest still decode).
  • array — a top-level JSON array → one record per element, decoded in a single pass. Error handling is atomic: a malformed array is dropped or fails as one payload (use ndjson when you need per-record isolation).

Typed or dynamic records​

  • build_serde::<T>() — decodes each document into your own T: serde::de::DeserializeOwned struct. No serde_json types leak into your pipeline. This is the flagship path — see crates/etl/examples/json_ndjson_memory.rs.
  • build_value() — decodes into dynamically-typed serde_json::Value records, for pipelines that inspect or route on structure not known at compile time.

Error handling​

A document that does not parse (or does not match the target type) is a record-level error, handled by on_error:

  • skip (default) — drop the record, count it in etl_json_deser_records_dropped_total{reason}, and continue. Under ndjson the good lines around a bad one still flow.
  • fail — surface a decode error on the first bad record.
on_error: fail and the chain's policy

fail makes the deserializer return a decode error; whether that stops the pipeline or drops the payload is then the chain's deserializer error policy (ErrorPolicy, default Skip). Set the chain's policy to Fail (.deser_error_policy(ErrorPolicy::Fail)) for a bad record to stop the pipeline. The common case — per-record skip — needs no chain-side change.

The connector-owned etl_json_deser_* families are minted when the builder is given a metrics scope with .with_metrics(pipeline, component); they sit alongside the framework's generic etl_deser_* stage metrics, which wrap every decoder. Undecodable data that stops the pipeline replays under at-least-once — replay never loses records.

Fidelity & features​

  • reject_duplicate_keys (config) turns serde_json's silent last-value-wins on duplicate object keys into a hard error — a guard against upstream corruption.
  • The float-roundtrip Cargo feature makes f64 values survive text→float→text exactly.
  • The raw-value feature exposes serde_json's RawValue for late/partial parsing (route on one field, forward the rest untouched).
arbitrary-precision

The arbitrary-precision feature preserves integers beyond u64 and exact decimals, but it is crate-wide once enabled and interacts with #[serde(flatten)], #[serde(untagged)], and RawValue. Enable it deliberately for a decimal/money pipeline, not by default.

By default, integers up to 64-bit are exact; values beyond that fall to f64 (the standard JSON interop limit). Duplicate keys are last-value-wins unless reject_duplicate_keys is set. NaN/Infinity are not valid JSON and are rejected.

Backends​

Decoding uses serde_json (stable 1.x) by default. The byte-slice → value step sits behind an internal seam, so the opt-in simd Cargo feature (json-simd on the etl facade) swaps in simd-json, a SIMD-accelerated parser, with no change to the builder API — build_serde and build_value are identical either way. from_reader is never used on the hot path — decoding always operates on the in-memory payload slice, which is both faster and lets the parser see the whole document at once.

simd decodes roughly 1.4× faster than serde_json on the single-document and array framings (the Kafka-message default), narrowing to ~1.15× on NDJSON — see the deserialization-formats benchmark for the full A/B and the aarch64 caveat. It is off by default and excluded from the facade's full feature: it is a per-deployment choice worth benchmarking on your own hardware and payloads. A note on how it works — simd-json parses a mutable buffer in place, so the framework copies each borrowed payload into a reused thread-local buffer first (measured negligible, ~1% of the decode); the reject_duplicate_keys guard always runs on serde_json, so duplicate-key handling is identical either way.

simd is not byte-for-byte identical to serde_json on numbers, though — it is a different parser. simd-json rejects integer literals outside the i64/u64 range (which serde_json accepts, coercing to f64), so a document serde_json decodes may instead be dropped as malformed — or fail the whole payload under on_error: fail — under simd; it also normalizes -0 to 0. And because parsing is done by simd-json rather than serde_json, the serde_json-specific knobs json-float-roundtrip, json-arbitrary-precision, and json-raw-value do not apply under simd (arbitrary precision is silently lost; RawValue capture is unsupported). For payloads of ordinary-range numbers the backends agree — but benchmark and validate against your own data before switching.