8 hours. That's how long a nightly batch job took when I picked it up two weeks ago. The team had already thrown a bigger cluster at it. Twice. It barely moved.
By the end it ran in 47 minutes. Same cluster. Same algorithms. I didn't touch the "smart" part of the pipeline at all.
The whole win was in how the data entered the system. The parser. The first few stages nobody ever profiles because they look boring.
This post is the long version of that fix — written so that if you've only been doing this for a year, you can follow every line. We'll define the jargon as we go, measure before we touch anything, and then walk through the five things that are almost certainly eating your runtime, each with before-and-after code you can run.
First, what is "the layer"? (30 seconds for newcomers)
A data pipeline is just a chain of stages. Raw data goes in one end, something useful comes out the other:
raw files -> parse -> transform -> aggregate / model -> output
-
Parse (also called ingestion or deserialization) means turning raw text — a JSON file, a CSV, a log line — into objects your program can work with, like Python dictionaries and lists. The bytes
{"id": 7}on disk become the dict{"id": 7}in memory. That conversion is not free. - Transform is where you clean, reshape, join, and compute features.
- Model / aggregate is the part people think of as "the real work" — the algorithm, the SQL, the math.
Here's the thing almost nobody internalizes early: that first box, parse, often costs more than all the others combined. And because it sits at the front, everything downstream pays its tax. A slow parser doesn't just slow down parsing — it slows down every stage that waits on it.
So when a pipeline is slow, the instinct is to make the smart parts faster. Usually the smart parts were never the problem.
Rule 1: never optimize what you haven't measured
Before you change a single line, find out where the time actually goes. This is the most important habit in this whole post, and it's three lines of code.
import time
from contextlib import contextmanager
@contextmanager
def stage(name):
start = time.perf_counter()
yield
print(f"{name:<18} {time.perf_counter() - start:6.2f}s")
Now wrap each stage of your pipeline in it:
import json
with stage("read + parse"):
records = [json.loads(line) for line in open("events.jsonl")]
with stage("transform"):
rows = [transform(r) for r in records]
with stage("aggregate"):
result = aggregate(rows)
You'll see something like this, and it's the whole story:
read + parse 372.10s
transform 48.30s
aggregate 12.05s
Parsing is 86% of the runtime. The "data science" everyone wanted to optimize is the bottom two lines.
When you need more detail — which function inside parsing is slow — reach for the profiler that ships with Python:
import cProfile
cProfile.run("run_pipeline()", sort="cumulative")
Read the top few rows of the output. If you see names like loads, decode, raw_decode, or scanstring near the top, that's the JSON parser, and you've just confirmed your bottleneck is ingestion — not the cluster, not Spark, not your SQL.
Everything below is what to do once the profiler points at parsing.
The five things quietly eating your night
1. You parse the same data five times
This is the most common one, and it hides because each stage looks reasonable on its own. The bug only shows up when you zoom out.
def clean_stage():
data = [json.loads(l) for l in open("events.jsonl")] # parse #1
return [r for r in data if r["status"] == "ok"]
def feature_stage():
data = [json.loads(l) for l in open("events.jsonl")] # parse #2 (!)
return build_features(data)
def report_stage():
data = [json.loads(l) for l in open("events.jsonl")] # parse #3 (!!)
return summarize(data)
Three functions, each independently re-reading and re-parsing the same file from scratch. If parsing is 80% of your cost, you just paid it three times. I've seen pipelines pay it seven or eight times across a DAG because each task was written by a different person on a different day.
The fix is embarrassingly simple: parse once, pass the result down.
records = [json.loads(l) for l in open("events.jsonl")] # parse ONCE
cleaned = clean_stage(records)
features = feature_stage(cleaned)
report = report_stage(features)
In a real orchestrator (Airflow, dbt, Spark), "parse once" usually means land the parsed data somewhere — a temp table, a cached DataFrame, a Parquet file — so the next task reads the cheap version instead of re-parsing the raw source. More on that in #5.
2. You hydrate the whole object to use three fields
Say each record is a big nested blob, but your job only ever touches three values.
def parse(line):
obj = json.loads(line) # builds the ENTIRE nested tree in memory
return obj # ...and you carry all of it downstream
You can't avoid parsing the line to find your fields — but you can avoid carrying the whole thing. Pull out what you use and drop the rest at the boundary:
import orjson # a C-based JSON parser, typically several times faster than the stdlib
def parse(line):
obj = orjson.loads(line)
return {
"user_id": obj["user"]["id"],
"ts": obj["event"]["timestamp"],
"amount": obj["event"]["payload"]["amount"],
}
Two wins here, and it's worth being precise about which is which:
-
orjsonswaps the slow pure-Python decode path for a fast C one. That's a real parse-time speedup, and it's a one-line change —pip install orjson, changejsontoorjson. - The field projection doesn't make the parse cheaper (you still parse the line). It makes everything after the parse cheaper: less memory held, smaller objects to shuffle and serialize between stages, and no repeated deep lookups like
obj["event"]["payload"]["amount"]happening over and over in later code. You do the deep walk once, here, and hand downstream a flat little dict.
Small objects move faster through every stage. A flat 3-key dict is cheaper to pass to Spark, pickle to a worker, or write to disk than a 200-key nested tree.
3. You load a 10 GB file just to read it once
If a single JSON document is bigger than your RAM, json.loads(f.read()) will either crash or swap to disk and crawl. You don't need the whole thing in memory at once — you need one record at a time. That's called streaming, and ijson does it:
import ijson
with open("huge.json", "rb") as f:
for record in ijson.items(f, "item"): # yields one array element at a time
process(record)
ijson walks the file as a stream and hands you records as it finds them, so memory stays flat whether the file is 100 MB or 100 GB. The "item" argument is the path to the elements you want — for a top-level JSON array [ {...}, {...} ], each element is item.
If your data is newline-delimited JSON (one object per line — the .jsonl format), you don't even need a library; just iterate the file line by line and parse each line, which is already streaming:
with open("events.jsonl") as f:
for line in f: # one line in memory at a time
record = orjson.loads(line)
process(record)
The lesson: never pull a whole dataset into memory to look at it piece by piece.
4. You let Spark guess your schema
This one is a Spark-specific tax that surprises a lot of people. When you do this:
df = spark.read.json("s3://bucket/events/")
Spark doesn't know the structure of your JSON, so before it can read the data it launches an extra pass over the files just to figure out the schema — what columns exist and what types they are. Then it reads everything again to actually load it. On a big dataset that inference pass is a full, expensive scan you're paying for silently, and on wide/nested JSON it can also blow up driver memory.
Tell Spark the schema up front and it skips the guessing pass entirely:
from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType
schema = StructType([
StructField("user_id", StringType()),
StructField("ts", LongType()),
StructField("amount", DoubleType()),
])
df = spark.read.schema(schema).json("s3://bucket/events/")
Same result, one less scan, lower memory, and a bonus: an explicit schema means a malformed record fails loudly instead of silently inferring the wrong type and poisoning everything downstream. (The same idea applies to spark.read.csv(..., inferSchema=True) — inference always costs an extra read.)
5. The real fix: JSON is the wrong shape to begin with
Everything above is treating symptoms. Here's the cause.
JSON is row-oriented and untyped text. Every time anything reads it, it has to be parsed again from scratch, and the types re-derived. There is no way to read "just the amount column" — to get one field out of every record, the parser still has to chew through every record. You pay the full parse cost on every single read, forever.
The fix is to pay that cost exactly once, at ingestion, and convert the raw JSON into a columnar, typed format. The standard choice is Parquet.
import pyarrow.json as paj
import pyarrow.parquet as pq
# At ingestion, ONCE: parse the JSON a single time and write Parquet
table = paj.read_json("events.jsonl")
pq.write_table(table, "events.parquet")
From then on, every downstream job reads the Parquet, and reading Parquet is a different universe:
import pyarrow.parquet as pq
# reads ONLY these two columns off disk; the rest is never touched
df = pq.read_table("events.parquet", columns=["user_id", "amount"])
Why this is so much faster, in plain terms:
-
Columnar means values for each column are stored together, instead of record-by-record. So "give me the
amountcolumn" reads one contiguous chunk and ignores the other 197 columns. In row-oriented JSON, the same request has to read and parse every record in full. - Typed means the column already knows it's a 64-bit integer. No text-to-number parsing on read. The types are stored in the file.
-
Compressed + indexed means Parquet keeps min/max stats per chunk ("row group"), so a filter like
amount > 1000can skip entire chunks without reading them. That's called predicate pushdown — the filter is pushed down into the read itself.
The mental model: JSON makes every reader redo the work. Parquet does the work once and lets every reader skip to exactly what it needs. If your raw data lands as JSON and gets read more than once — and it always does — converting it to columnar at the front door is usually the single highest-leverage change you can make.
The pattern behind all five: normalize once at the front door
Step back and every fix above is the same move. Do the expensive, messy work — parsing, type-casting, field extraction, schema enforcement — once, at the boundary where data enters your system. Produce one clean, typed, columnar version. Then let every downstream stage read that cheap version instead of fighting the raw input over and over.
raw JSON -> [ parse + flatten + type + write Parquet ] -> clean columnar table
(pay this ONCE) |
v
every model / report / job reads this
This is why the 8-hour job became a 47-minute job without touching the algorithms. The old version re-parsed raw nested JSON inside every stage, inferred the schema on every read, and dragged the full objects through the whole DAG. The new version parsed once, landed typed Parquet, and let everything downstream read flat typed columns. Nothing got smarter. The work just moved to the layer it belonged in — and stopped being repeated.
The principle
If your parser is wrong, every layer downstream pays for it. Your model is only as fast as the data feeding it, and your data science is only as good as the shape the data arrives in. You can have a perfect algorithm sitting behind a parser that re-reads 10 GB of JSON five times a night, and you will spend your week buying bigger clusters to fix a problem that lives in the first ten lines of the job.
So before you reach for more compute, profile the thing and find out where the time actually goes. Nine times out of ten it's hiding at the front door, in the boring stage nobody looks at.
Last time you chased a slow pipeline — when you finally measured it, where did the time actually turn out to be hiding? I collect these stories.
I'm Vinicius Fagundes — principal data engineer, independent, and an MBA lecturer in São Paulo. I spend my days making slow, expensive, fragile pipelines fast, cheap, and boring. If this sounds like your stack, this is the work I do at vf-insights.com.
