In Apache Beam, when does a large side input become the bottleneck in a ParDo?

## What a side input is Every `PCollection` a `ParDo` reads element-by-element is its *main input*. A **side input** is an additional `PCollection` that the transform can consult *in its entirety* on every element. In Python you wrap it at the call site: ```python rates = (p | "ReadRates" >> beam.io.ReadFromText("gs://ref/fx.csv") | "Parse" >> beam.Map(parse_kv)) converted = (orders | "Convert" >> beam.Map( lambda order, fx: {**order, "usd": order["amount"] * fx[order["ccy"]]}, fx=beam.pvalue.AsDict(rates))) ``` The wrappers describe the view your function receives: - `AsSingleton` — the collection must contain exactly one element; you get that value. - `AsIter` — an iterable over the elements. - `AsList` — a real list, fully materialized. - `AsDict` — a dict built from `(key, value)` pairs; keys must be unique. Conceptually this is a **broadcast join**: the small side is replicated everywhere so the big side never has to be shuffled. ## Why size is the whole question Three costs scale with the side input. **Materialization.** The side input is itself the output of a subgraph. Before the consuming `ParDo` can run, that subgraph must complete and its result be written somewhere the workers can read. **Distribution.** Every worker that runs the consuming `DoFn` needs access to the entire side input, not a slice of it. Total work is roughly side-input size times number of workers, which is exactly the term that grows when a job autoscales up. **Worker memory.** `AsList` and `AsDict` construct the whole structure in the worker process. A side input that comfortably fits on your laptop under `DirectRunner` can be the reason a distributed job dies with an out-of-memory error, and the failure mode is confusing because the *main* input looks innocent. `AsIter` avoids building one giant Python object but still requires access to the full collection. There is no published size at which this flips, and you should not quote one in an interview. The honest answer is a shape rule: side inputs are for reference data small enough to sit comfortably in worker memory alongside your other state; once the lookup side approaches the scale of the main input, use a shuffle-based join. ## The barrier property A side input is also a synchronization point. The consuming `ParDo` cannot emit results for a window until the side input for that window is ready. In batch this means the side-input branch gates the main branch — an expensive side-input computation delays everything downstream of it. In streaming, side inputs are **per window**: as the main input advances into a new window, the runner needs the side-input view for that window, and main-input elements wait until it is available. A side input derived from a slowly updating reference stream is a common source of "my streaming pipeline stalls periodically" reports. ## The alternative: CoGroupByKey When both sides are large, the correct primitive is a shuffle join: ```python joined = ({"orders": keyed_orders, "customers": keyed_customers} | beam.CoGroupByKey()) ``` `CoGroupByKey` groups several keyed `PCollection`s by the same key and hands you, per key, the values from each tagged collection. Both sides are shuffled — more network than a broadcast — but nothing has to fit on a single worker, and the cost scales with data rather than with data times worker count. The trade is the classic one: broadcast join when one side is small, shuffle join when neither is. ## Diagnosing it in practice Symptoms that point at a side input: - Workers OOM in a step that does trivial per-element work. - The job stalls before the consuming step even starts, while an unrelated-looking branch is still running. - Scaling the job up makes it *worse*, because every new worker must pull the whole side input again. - A streaming pipeline's latency spikes at regular intervals matching the side-input window. Fixes, in order of preference: shrink the side input (project only the columns you need, filter to the keys that actually occur, pre-aggregate it); switch `AsList` to `AsIter` or `AsDict` keyed lookups so you are not holding a list you scan linearly; move to `CoGroupByKey`; or, for genuinely huge reference data with point lookups, query an external key-value store from `setup()` in the `DoFn` and accept the per-lookup latency. ## What to say "A side input is Beam's broadcast join — the whole collection is visible to every `DoFn` call, wrapped with `AsDict`, `AsList`, `AsSingleton` or `AsIter`. It is great for small reference data and bad when it is large, because it is materialized, copied to every worker, and with `AsList`/`AsDict` held in worker memory, and because it gates the consuming step until it is ready per window. Once both sides are big I use `CoGroupByKey` instead."