ahmedabu98
commented
3 days ago I ran two identical pipelines (writing 10B records) before and after this change to measure the difference (this is in a project with large quota):
Job before fix (2024-06-27_22_27_46-11409968544737429803)
- 330 workers
-
Creates 1,950 connections and never closes them (they eventually time out).
- ~290 GiB append rows throughput
- Finishes in 22.5 min
- Dataflow cost estimate: $10.00
Job after fix (2024-06-27_22_53_21-1016146269936299010):
-
283 workers
-
Up to 550 connections at a time. All connections get cleaned up before the pipeline is finished
-
~250 GiB append rows throughput
-
Finishes in 21.5 min
-
Dataflow cost estimate: $6.93
Side by side comparison (the jobs were run sequentially):
There is a significant reduction (roughly ~70%) in concurrent connections. We can reliably expect the number of concurrent connections per destination to be capped at the number of parallel DoFns (or vCPUs). In other words, concurrent connections <= (num vCPUs) x (num destinations)
github-actions[bot]
Abacn
reuvenlax
liferoad
Storage API connections in batch are left open and not closed properly. This is because we pin the underlying StreamAppendClient twice: once for the bundle and once for the cache When we are finished with the stream however, we only unpin once for the bundle (and not for the cache).
Batch mode already creates a lot of streams and connections (one stream/connection per destination per bundle) compared to streaming mode. Leaving connections unclosed leads to many concurrent connections and can quickly exhaust the quota.
This change adds a line to invalidate the cached client after we finish using it in a bundle.
Also creates a counter to keep track of active connections.