In an Apache Beam DoFn, what runs in setup versus start_bundle versus process?

## Where the lifecycle sits `ParDo` is Apache Beam's general-purpose element-wise transform; `beam.Map`, `beam.FlatMap` and `beam.Filter` are conveniences built on top of it. When you need per-element work with real setup, side outputs, or state, you write a `DoFn` subclass and pass it to `beam.ParDo`. The runner does not call your `DoFn` once per element in isolation. It creates instances on workers, groups elements into **bundles**, and drives a nested lifecycle around them. ## The five hooks, outermost first **`__init__`** runs in your *driver program*, on your machine, when you build the graph. Whatever you assign here is serialized (pickled, in Python) and shipped to workers. That is why you must not create a database client or an open socket in `__init__` — it will not serialize, or it will serialize into something dead. Constructor arguments should be plain configuration values. **`setup()`** runs once per `DoFn` instance, on the worker, before any element is processed. This is the correct home for expensive, reusable initialization: an API session, a connection pool, a loaded model file, a compiled regex table. Because one instance serves many bundles, this cost is amortized. **`start_bundle()`** runs once before the elements of a bundle. A bundle is a runner-decided group of elements — its size is not something you control or should assume. Use `start_bundle()` to reset per-bundle mutable state, typically an output buffer. **`process(element)`** runs once per element and returns or yields zero, one, or many outputs — that is why `ParDo` generalizes both map and filter. Extra arguments can be injected: `beam.DoFn.TimestampParam`, `beam.DoFn.WindowParam`, and side inputs passed by the caller. **`finish_bundle()`** runs after the last element of a bundle. Flush whatever `process()` buffered. A sharp edge: if you *emit output* from `finish_bundle()`, you must emit a `WindowedValue` with an explicit timestamp and window, because there is no current element to inherit them from. Most people only flush side effects here and emit nothing. **`teardown()`** is called when the runner discards the instance — close the connection you opened in `setup()`. It is explicitly **best effort**. A worker that is killed, preempted, or crashes will never call it. Never rely on `teardown()` for correctness, only for tidy resource release. ```python class PostToApi(beam.DoFn): def __init__(self, endpoint): self.endpoint = endpoint # plain config, gets serialized def setup(self): self.session = make_session() # once per instance, on the worker def start_bundle(self): self.buffer = [] def process(self, element): self.buffer.append(element) yield element def finish_bundle(self): if self.buffer: self.session.post(self.endpoint, json=self.buffer) def teardown(self): self.session.close() # best effort only ``` ## Bundles are the retry unit — and that is the real interview point If any element in a bundle fails, the runner re-runs **the whole bundle**, calling `start_bundle()`, `process()` on every element again, and `finish_bundle()` again. Consequences: - Side effects in `process()` or `finish_bundle()` can happen more than once. An external write must be idempotent — an upsert keyed by a stable id, or a write whose duplicate is harmless — or you will double-count on the first transient failure. - Counters you keep in Python variables are wrong under retries; use `beam.metrics.Metrics.counter` if you want observable counts, and treat even those as approximate. - State that survives across bundles in instance attributes is dangerous. Anything set in `setup()` should be read-only or a reusable resource; anything accumulating results belongs in `start_bundle()` so a retry starts clean. ## Common bugs this question is really probing 1. **Opening a connection per element.** Putting client construction in `process()` is the classic performance defect; it belongs in `setup()`. 2. **Opening a connection in `__init__`.** Unserializable, and a source of "works locally, fails on the service" errors. 3. **Mutating the incoming element.** Beam requires input elements to be treated as immutable; mutating one breaks retry correctness and the `DirectRunner` will flag it. 4. **Assuming bundle size.** You cannot ask for batches of exactly 500 by relying on bundles. If you need a controlled batch size, use `beam.BatchElements` or `GroupIntoBatches`, both of which are real Beam transforms designed for that. 5. **Relying on `teardown()` to flush data.** Flush in `finish_bundle()`; `teardown()` only releases resources. ## What to say "`__init__` on the driver and gets pickled, `setup()` once per instance on the worker for expensive clients, `start_bundle()` per bundle to reset buffers, `process()` per element, `finish_bundle()` to flush, `teardown()` best effort. The bundle is the retry unit, so every external side effect in there has to be idempotent."