Pipeline compute

Centrum and the macro execution pipeline run on classic compute. Everything about it is declared in infrastructure/env/{dev,prod}/main.tf: a cluster policy per pipeline pins the node types and the spot settings, and the pipeline module sets the worker counts, the edition and the mode. The metrics pipeline is serverless and has no node types to choose.

This page is the method. The current configuration, and why it beat the alternatives, is recorded in the pipeline compute ADR.

How to choose

Work through these in order. Each step narrows the candidates before the next one compares prices.

1. Size for the code that will run. Measure the architecture the new configuration will carry. Production may still run older code, and dev may be ahead of it or behind it. Write down which one you measured and in which direction it errs.
2. Measure on long-lived clusters. Memory builds up over days, so a cluster that restarts every hour looks healthier than it is. The query at the end of this page reads system.compute.node_timeline per pipeline.
3. Read memory as swap, not as memory used. Spark claims its heap when it starts, so memory used is always high and says nothing about headroom. Swap shows that a node is full. I/O wait alongside it suggests the node is paging, but disk and network work raise I/O wait too, so it only counts together with swap.
4. Classify each role. A worker is CPU-bound, memory-bound, or waiting on something outside the cluster. The driver runs no tasks: its memory grows with the number of flows and the tasks it tracks, and its CPU plans every micro-batch.
5. Measure each flow, not only each node. Node averages hide a cost that every micro-batch pays whatever it carries. Spark's REST API on a running pipeline cluster, /driver-proxy-api/o/<workspace id>/<cluster id>/40001/api/v1/applications, gives the tasks and executor time of each flow's stages. A flow that runs hundreds of tasks for a few rows is paying a fixed charge on every trigger, and it delays everything that waits for a free slot, including the Kinesis reader.
6. Size streaming shuffles to the cores. Stateful streaming operators, such as deduplication, skip adaptive execution, so their shuffles run at spark.sql.shuffle.partitions, 200 by default, however small the batch. Check a flow's stages or query plan before sizing: a stage with 200 tasks for a handful of rows is the sign. Set it in the pipeline cluster's Spark configuration, to the cores the workers can reach; the flows inherit it. Auto compaction is the exception: it plans every write in its own session at 200 partitions, whatever the pipeline or the cluster sets. In a continuous pipeline, turn it off on every table, materialized views included, with delta.autoOptimize.autoCompact and let predictive optimization compact the tables asynchronously, after checking that predictive optimization is enabled for the catalog. Set a stateful operator's partitions on its own table with spark.sql.streaming.stateStore.partitions in the table's spark_conf, sized for its state rather than the cluster: it keeps the count its checkpoint was created with unless Databricks repartitions the state, and otherwise only a full refresh changes it.
7. Fix memory per core first. Python UDF workers run outside Spark's heap, one per running task, and they cannot spill to disk. More cores on the same memory means more of them. Memory per core decides the family: r gives 8 GB per core, m 4 GB and c 2 GB.
8. Then pick the generation. DBUs are most of a pipeline's bill, and Databricks raises the DBU rate per core by about 10% with each generation. AWS's claims of better price-performance per generation are measured per EC2 dollar, so they do not settle the choice. Decide on how often spot capacity is reclaimed in eu-central-1, whether a local NVMe variant exists and the workspace runs the disk cache on it, and on core speed only where the role is CPU-bound.
9. Then pick the size. A smaller size is often reclaimed less and costs the same or less per core. A larger one carries less fixed overhead per node but moves autoscaling in bigger steps.
10. Count waiting work as slots. When tasks wait on HTTP, a core is a task slot, not processing power. Size the slots against the limit downstream: for macros, slots times requests in flight per task must stay within the sandbox's reserved concurrency.
11. Keep drivers on-demand, without local NVMe, and on at least four cores. Losing the driver stops the update, and a driver never uses local disk. On two cores the Spark driver often misses its 300-second startup timeout, so a restart often takes more than ten minutes or fails. Size its memory for the flows and its cores for the peaks.
12. Let the features decide the edition. Expectations require ADVANCED, which Centrum still uses. Centrum's bridge flows are append-only and no longer use AUTO CDC. The macro pipeline has neither expectations nor AUTO CDC and runs on CORE.
13. Keep spot for workers only. Workers run on spot with fallback to on-demand, behind an on-demand driver. Production Centrum carries ingest, so its first worker runs on-demand too (first_on_demand = 2): one reclaim can take every spot worker at once, and when it did, every flow stopped for 14 minutes, bronze included. The on-demand worker keeps the flows running while replacements start, for about $85 a month. The pipeline sets this in its own aws_attributes as well as in the policy, because a policy's fixed values are not documented to reach pipeline clusters. Dev and the macro pipeline accept the stall, because a few minutes of delay there costs less than an on-demand node.
14. Roll out on dev first and gate production on the code. Watch dev for at least a day. Terraform applies before the bundle deploys, and a continuous pipeline restarts by itself when its settings change, so it comes back on the new compute running its old code. Restart it again once the bundle has deployed. If the apply also adds notebooks to the pipeline, that first restart fails until the bundle uploads them. Production must not get compute sized for code it is not yet running.

Macro batch limits

The macro pipeline sets three controls in its Terraform configuration:

The reader stops admitting files when either input limit is reached. These are starting values to tune on dev, not guarantees of row count, memory use or elapsed time. A single oversized file can exceed the byte budget. Exploding macros and expanding their results can increase the work further. Range partitioning spreads distinct row IDs within each macro/version group; identical keys can remain skewed, and small batches need not fill every partition.

Each task retains up to three concurrent HTTP requests. Each request contains one macro/version group, at most 25 items and the existing HTTP byte budget. Imported and invalid macro rows skip the sandbox inside the Python UDF. Transient request retries remain enabled, so sandbox execution is not exactly-once.

Admission settings can change while retaining the existing source, flow, table and checkpoint. An already planned batch must finish with its original input range before the new limits apply. Restarting with smaller input limits will not reduce that batch's input range. The stateless range shuffle and eligibility guard do apply when an unfinished batch is replayed, so they can reduce task size and unnecessary sandbox calls within that same range. Use a normal checkpoint-resuming update, never a full refresh or manual offset edit. See Databricks admission controls.

On dev, verify a completed batch and an interrupted batch resume without missing or duplicated committed rows. Watch batch duration, commit gaps, maximum task rows, backlog growth and backend errors before promoting to production. A shorter commit gap is insufficient if the backlog keeps growing. Watch the backend WAF rate-rule blocked-request metric and macro errors containing Chunk failed and 403; a WAF rejection becomes a failed result without retry. Restore the previous settings to roll back; the same pending-batch constraint applies.

Reading a driver's memory

Both classic pipelines deliver their driver and worker logs to the pipeline-logs volume in the centrum schema, in a folder per pipeline and then per cluster. Without delivery, a driver's logs are gone when its cluster ends.

The driver's stdout carries the JVM's GC log. Each full collection logs Pause Full ... A->B, where B is the heap still in use afterwards. If B climbs for hours, the driver is keeping memory. Once it reaches the old generation's size, the driver spends most of its time collecting, and the pipeline's event log reports gc_pressure. The heartbeat export reads the newest driver's log every half hour and publishes how full the old generation is after the latest full collection, per pipeline. The driver-heap alert fires above 80%, which in September came about eleven hours before Centrum's driver stalled. Setting JAVA_TOOL_OPTIONS=-Xlog:gc+classhisto=trace in the cluster's environment variables adds a class histogram at every full collection, which shows what that memory is. Each histogram lengthens its collection, so remove the setting once it has answered the question.

Where the numbers come from

A node costs its DBU rate times the price per DBU, plus the EC2 price, per hour.

Utilization query

SELECT t.driver,
  round(avg(t.cpu_user_percent + t.cpu_system_percent), 1) AS cpu_avg,
  round(percentile(t.cpu_user_percent + t.cpu_system_percent, 0.95), 1) AS cpu_p95,
  round(avg(t.cpu_wait_percent), 1) AS io_wait_avg,
  round(avg(t.mem_swap_percent), 1) AS swap_avg,
  round(max(t.mem_swap_percent), 1) AS swap_max
FROM system.compute.node_timeline t
JOIN (
  SELECT DISTINCT cluster_id FROM system.compute.clusters WHERE cluster_name LIKE '%<pipeline id>%'
) c USING (cluster_id)
WHERE t.start_time >= current_timestamp() - INTERVAL 7 DAYS
GROUP BY t.driver

Known gaps

• Graviton and fleet instance types are offered in the workspace, but Databricks publishes no pipeline price for them, and pipeline support for them is unverified.
• Photon raises the DBU rate and does not speed up Python UDFs, which most of our flows use.
• A policy that fixes a node type has to change in the same apply as the pipeline that names it. A pipeline that disagrees with its policy fails at its next start.