JSONL → Parquet+ZSTD Converter (v2)

S3-to-S3 streaming converter that compresses JSONL files into Parquet with ZSTD compression. Built and tuned against the viraj/ and code-gen/ prefixes on bgen-data-team. Production-tested on 51.77 TiB → 8.54 TiB (83.5% smaller).

Two converters in this repo

• convert_jsonl_parquet.py — streaming S3-to-S3 via DuckDB httpfs (this README).
• pipeline_convert.py — 3-stage download→convert→upload via local NVMe. DuckDB is local-only; boto3 owns S3. Reach for it when token expiry, httpfs flakiness, or huge single-file converts bite. See PIPELINE.md.

Input  (S3 JSONL)  ──┐                              ┌──> Output (S3 Parquet+ZSTD)
                      ├── DuckDB stream + COPY ──> ──┤
Manifest (S3)  ──────┘                              └──> Progress (local JSON)

Production results

Metric	Value
Source prefix	s3://bgen-data-team/viraj/
Files converted	1,570
Input size	51.77 TiB
Output size	8.54 TiB
Compression ratio	83.5% smaller
Bytes saved	43.23 TiB
Failures	0
Format	Parquet + ZSTD-9
Row group size	1,000,000 rows
Hardware	191 cores, ~1.2 TiB usable RAM

---

Quickstart

# Single-node, default settings (good for code-gen-shaped data)
python3 convert_jsonl_parquet.py \
  --src-prefix viraj/ \
  --num-nodes 1 --node-id 0 \
  --large-workers 1 \
  --threads-per-large-worker 64 \
  --memory-per-large-worker 200GB \
  --small-workers 20 \
  --memory-per-small-worker 32GB \
  --size-threshold-gb 0.5 \
  --compression-level 9 \
  --allow-malformed

# Dry-run (no writes, just shows what would convert)
python3 convert_jsonl_parquet.py --src-prefix viraj/ --dry-run

# Resume an interrupted run — just re-run the same command.
# Already-completed files skip via progress.json + footer validation.

# Retry failed files (with patched script after fixes)
python3 convert_jsonl_parquet.py ... --reset-failed --allow-malformed

Run under tmux or nohup for long jobs.

---

How it works

Streaming pipeline (no local disk)

DuckDB reads JSONL directly from S3 via httpfs, parses, compresses to Parquet, and writes back to S3 — all in one streaming COPY statement. No intermediate files. Memory caps prevent unbounded growth.

Output key transform

Input:  s3://bucket/<src>/path/to/file.jsonl
Output: s3://bucket/<src>-parquet/<2hex>/path/to/file.parquet
                                  ^^^^^^^
                                  md5(input_key)[:2]

The 2-hex prefix spreads writes across 256 sub-prefixes — sidesteps S3's 3,500 PUT/sec-per-prefix throttle without affecting reads.

Two-phase worker pools

Pool	When	Workers	Memory each	Threads each
Large (Phase 1)	file ≥ --size-threshold-gb	few (1)	large (200 GB)	many (64)
Small (Phase 2)	file < --size-threshold-gb	many (20)	small (32 GB)	few (auto)

Huge files need lots of memory; tiny files parallelize wide.

Multi-node partitioning

Bin-packs by bytes (not file count) for balanced node load. Deterministic across nodes: same input → same partition assignment.

# Node 0 (also lists + writes the shared S3 manifest)
python3 convert_jsonl_parquet.py --num-nodes 3 --node-id 0 ...

# Node 1 / 2 (read manifest from S3, process their slice)
python3 convert_jsonl_parquet.py --num-nodes 3 --node-id 1 ...
python3 convert_jsonl_parquet.py --num-nodes 3 --node-id 2 ...

Start node 0 first so the manifest is built before others try to read it.

Resume semantics

Two layers guard against re-conversion:

1. progress_<shard>.json records completed files; listing-stage filter skips them instantly.
2. Footer validation — before any COPY, the script HEADs the output and reads the Parquet footer. Valid footer → skip. Invalid → delete and reconvert.

So even if progress.json is lost, no work is duplicated.

---

Data-quality fallback chain

Real JSONL data has surprises. The script auto-handles three classes of issue:

1. Schema variation across rows → union_by_name = true

If row 3.18M of a file has a key not seen in the first 20K-row sample (e.g. extra "Subject" key), the typed read normally fails. union_by_name=true (always on) extends the schema as new keys appear instead of locking to the sample.

2. Malformed JSONL lines → --allow-malformed

For files like nemo-pt-v1/output2/passed/* that contain stray bad lines (e.g. concatenated JSON or trailing garbage). Sets ignore_errors=true — those rows are silently dropped, the rest of the file converts. Trade-off: row-count verification can't detect dropped rows.

3. Type-coercion errors → 3-tier auto-fallback

When DuckDB's 20K-row sample mistypes a column (e.g. inferred INT but later row has a string), the COPY blows up with a stoi/Conversion Error mid-file. The retry wrapper transparently tries deeper modes:

Tier 1: typed read (default)                       — best case, full structure
   │
   └─ TYPE error ──> Tier 2: sample_size=-1        — scan whole file for types
                       │                              mixed columns become VARCHAR
                       │
                       └─ TYPE error ──> Tier 3: read_json(records=false)
                                          single JSON-typed column called `data`
                                          downstream uses json_extract()

Success log surfaces the tier:

• OK file.jsonl ... → typed (tier 1)
• OK [FULL-SAMPLE] file.jsonl ... → tier 2
• OK [RAW-TEXT] file.jsonl ... → tier 3 (single-column)

To skip the retry-chain delay on known-problem files, force a tier directly with --full-sample or --raw-text.

---

Critical configuration constraints

These are load-bearing — change at your peril.

maximum_object_size must stay < INT32_MAX (2,147,483,647)

DuckDB parses this parameter via std::stoi (32-bit). Setting it to 2^31 (2,147,483,648 = 2 GiB exactly) throws bare "stoi" from SQL parameter parsing — looks identical to a data-side type error. Current value: 1 GiB (1,073,741,824).

--large-workers 1 (sequential, not parallel)

Parallel COPYs at this DuckDB version + host combo segfaulted. Single-worker is the only stable mode for the large-file phase.

http_keep_alive left at DuckDB default

Enabling it caused segfaults at 8-worker concurrency in earlier testing. Single-worker mode could safely turn it on but currently doesn't.

ZSTD level 9, not 15

Level 9 hits the sweet spot of size-vs-time for warm-storage workloads (~2-3 reads per few months). Level 15 added ~3-5% better compression at ~3× the CPU time — not worth it for read-rare data.

memory_per_large_worker must fit in (available RAM) / large-workers

DuckDB's memory_limit is a soft cap; real usage spikes past during schema inference + ZSTD buffers. Always leave 300-400 GiB headroom.

---

CLI reference

Flag	Default	Purpose
--bucket	bgen-data-team	S3 bucket
--src-prefix	code-gen/	Source prefix to scan
--dst-prefix	derived (<src>-parquet/)	Destination prefix
--compression	zstd	zstd or snappy
--compression-level	15	zstd 1-22 (use 9 for production)
--row-group-size	1,000,000	Rows per Parquet row group
--num-nodes	1	Total nodes in run
--node-id	0	This node's index (0-based)
--force-refresh-manifest	off	Node 0 only: rebuild S3 manifest
--max-files N	unlimited	Process at most N files
--min-size-gb / --max-size-gb	none	Size filters
--size-threshold-gb	1.0	Files ≥ → large pool
--large-workers	8	Workers for large-file phase
--small-workers	100	Workers for small-file phase
--threads-per-large-worker	96	DuckDB threads per large-worker
--memory-per-large-worker	100GB	DuckDB memory_limit per large-worker
--memory-per-small-worker	4GB	DuckDB memory_limit per small-worker
--allow-malformed	off	ignore_errors=true; drops bad JSONL lines
--full-sample	off	Force sample_size=-1 from tier 1
--raw-text	off	Force records=false (single-column)
--reset-failed	off	Clear failed list, retry those files
--dry-run	off	Show planned work, no writes
--progress-file	auto	Override progress JSON path

---

Monitoring an in-flight run

# Live log tail
tail -f conversion.log

# Snapshot of progress.json
python3 -c "
import json
d = json.load(open('progress_viraj.json'))
s = d['stats']
print(f'OK:    {len(d[\"completed\"])}')
print(f'FAIL:  {len(d[\"failed\"])}')
print(f'Saved: {(s[\"total_input_bytes\"]-s[\"total_output_bytes\"])/1024**4:.2f} TiB')
"

# Process state (is it actually doing work?)
ps -p $(pgrep -f convert_jsonl_parquet | head -1) -o pid,pcpu,pmem,etime,stat,comm

# Memory / spill check
du -sh tmp_duckdb/                       # spill directory size
free -h                                  # total + available RAM

# Audit: which files used the type-fallback?
python3 -c "
import json
d = json.load(open('progress_viraj.json'))
for k, v in d['completed'].items():
    if v.get('used_full_sample') or v.get('used_raw_text'):
        tag = 'RAW-TEXT' if v.get('used_raw_text') else 'FULL-SAMPLE'
        print(f'[{tag}] {k}')
"

---

Troubleshooting

stoi error on every file (fast, 30-60s each)

Cause: maximum_object_size ≥ INT32_MAX. The error is in SQL parameter parsing, not data. Fix: cap at 1 GiB (1073741824). Current code already has this.

Killed with no further output

Cause: OOM killer. Your large-workers × memory-per-worker exceeds available RAM. Note memory_limit is a soft cap — real usage spikes past it during schema inference + compression.

Fix: Check free -h, compute (available RAM - 400 GiB headroom) / (workers × memory). Reduce one of those. On the verified production host (2 TiB total, 750 GiB consumed by other workloads, 1.2 TiB usable): stay under ~800 GB total cap across all workers.

unknown key "X" schema error

Cause: A rare key appears later than DuckDB's inference sample. Fix: Current code treats this like schema drift and retries that file with sample_size=-1. If the schema still cannot stabilize, it falls back to raw JSON-line parquet. For a shard where most files have high-cardinality dynamic keys, start the run with --full-sample or --raw-text.

maximum_object_size of N bytes exceeded

Cause: A single JSON record is larger than the cap. Fix: Edit the constant in the script. Current value 1 GiB; do not go ≥ INT32_MAX.

Out of Memory during a small-pool file conversion

Cause: union_by_name=true schema state grew very large on a high-cardinality-key file. Fix: Bump --memory-per-small-worker to ≥32 GB, lower --size-threshold-gb to 0.5 so larger files move to the high-memory pool.

ExpiredToken mid-conversion

Cause: STS token expired (default 1 hour). The script has automatic secret refresh every 20 minutes, but a single file COPY taking >1 hour will fail.

Fix: Either split huge files, run with a 12-hour token, or accept the retry path (the script retries transient errors with fresh creds).

Segfault under high concurrency

Cause: Parallel COPYs at high --large-workers combined with http_keep_alive=true historically crashed DuckDB. Fix: Stick with --large-workers 1 for the large phase.

File hangs for many minutes with no output

It's almost certainly slow, not stuck. A 2 TiB file at 1 GB/s S3 read takes ~35 minutes just to download. The script doesn't log intra-file progress. Diagnose with ps, /proc/PID/io, du -sh tmp_duckdb/.

---

Output layout on S3

s3://bgen-data-team/
├── viraj/                                # input (untouched)
│   └── synthetic-code-pt/.../file.jsonl
│
├── viraj-parquet/                        # output
│   ├── 00/
│   ├── 01/
│   ├── ...
│   ├── a3/synthetic-code-pt/.../file.parquet
│   └── ff/
│
└── _manifests/
    └── viraj.json                        # shared file listing

Querying the output

-- Everything (all 256 shards):
SELECT * FROM read_parquet('s3://bgen-data-team/viraj-parquet/*/**/*.parquet');

-- A specific path (still globbed across shards):
SELECT * FROM read_parquet(
  's3://bgen-data-team/viraj-parquet/*/synthetic-code-pt/.../file.parquet'
);

-- Files written via [RAW-TEXT] fallback have a single `data` column:
SELECT json_extract(data, '$.field_name') FROM read_parquet(...);

---

File structure

v2/
├── convert_jsonl_parquet.py    # the script (everything)
├── delete_completed_inputs.py  # delete inputs after verifying outputs
├── conversion.log              # append-only run log
├── progress_<shard>.json       # resume state per source prefix
├── tmp_duckdb/                 # DuckDB spill directory (auto-created)
└── README.md                   # this file

---

Reclaiming source storage (delete_completed_inputs.py)

After conversions land cleanly, the JSONL inputs can be removed to reclaim S3 storage. delete_completed_inputs.py walks every progress_*.json next to it and, for each completed entry, deletes the input only after re-verifying the output on S3.

Per-file checks (all must pass)

1. HEAD the recorded output_key — must exist with ContentLength > 0.
2. Read the Parquet footer via DuckDB's parquet_file_metadata() — same check convert_jsonl_parquet.py does before skipping a pre-existing output. Fails on truncated / corrupt files.
3. (Optional --strict-row-count) the footer row count must equal the footer_rows value recorded at conversion time.

Only when all three pass does the script delete the input_key. The default is a dry-run — --execute is required to actually delete anything.

Usage

# 1) Dry-run — verifies every completed parquet, deletes nothing.
python3 delete_completed_inputs.py

# 2) After the dry-run is clean, actually delete.
python3 delete_completed_inputs.py --execute

# Stricter: also require footer rows to match the value recorded in progress.
python3 delete_completed_inputs.py --strict-row-count --execute

# Quick smoke test on the first 50 entries (verbose prints each one).
python3 delete_completed_inputs.py --max-files 50 --verbose

Flags

Flag	Default	Purpose
--bucket	bgen-data-team	S3 bucket for both verify and delete
--progress-file	every progress_*.json next to the script	Repeatable; explicit progress files to read
--execute	off (dry-run)	Without this flag, nothing is deleted
--strict-row-count	off	Also require footer rows == recorded footer_rows
--workers	32	Concurrent verifier/deleter threads
--max-files N	unlimited	Process at most N entries (for smoke tests)
--verbose	off	Log every entry considered, not just failures

Output

Per-file lines plus a summary at the end:

=== summary ===
Verified OK:    1,588
Verify failed:  0
Deleted:        1,588          # only in --execute mode
Delete failed:  0              # only in --execute mode
Reclaimed:      51.77 TiB
Elapsed:        ...

Verify failures are listed (first 20) with their reasons — these inputs are kept. Re-run after addressing them.

Safety notes

• Dry-run is the default — there is no flag that bypasses verification.
• Deletes are driven by progress_*.json, not by listing the destination prefix. A parquet that happens to exist on S3 but isn't recorded in progress will not cause its input to be deleted.
• The footer check is the same one the converter uses to decide a pre-existing output is trustworthy. If it passes here, the parquet would also be trusted on a future resume.
• --strict-row-count is the strongest setting; use it unless you're re-converting files where --allow-malformed legitimately dropped rows (row counts won't match in that case).

---

Lessons learned (the journey, briefly)

1. DuckDB's read_json_auto doesn't accept all_varchar (only read_csv_auto does). Tier-3 fallback first used a broken parameter, then read_csv with control-char delimiters as a hack, then finally the correct read_json(records=false).

2. maximum_object_size = 2^31 is the worst possible value — exactly one too high for stoi. Always cap below INT32_MAX.

3. Auto-fallback chains are good defaults when data has long-tailed quality issues. Manual flags exist for when you've already learned a file needs tier N.

4. S3 prefix throttling is real but --large-workers 1 makes it moot at this scale. The 2-hex sharding was still worth keeping for safety margin and zero read penalty.

5. DuckDB memory_limit is soft. Plan for 1.5-2× the cap during peak moments (schema inference + ZSTD buffer + row-group flush).

6. STS auto-refresh in DuckDB doesn't work via CREDENTIAL_CHAIN. Resolve via boto3 in Python, push frozen creds into DuckDB secrets, refresh every 20 min explicitly.

7. Verify hardware claims with free -h and htop. Stated specs and reality often diverge by 2-4× on shared hosts.

s3-compression