Uber is dedicated to offering dependable companies to clients throughout our world markets. To attain this, we closely depend on machine studying (ML) to make knowledgeable selections like forecasting and surge. In consequence, real-time streaming pipelines, that are used to generate the information and options for ML, have turn out to be extra widespread and necessary.
At Uber, we leverage Apache Flink to construct the real-time streaming pipelines, and construct platforms like Gairos and AthenaX to simplify growth. Nevertheless, there are nonetheless many challenges, reminiscent of scalability, as a result of both the complexity of computation or the quantity of real-time information to be processed.
On this article, we are going to use the pipelines that generate demand and provide options, for example to introduce a few of the challenges we confronted and the way we solved them. Particularly we are going to clarify how we tune the real-time pipelines with the efficiency tuning framework.
The determine under reveals the high-level structure: Streaming Pipelines in Apache Flink are accountable for the characteristic computation and ingestion. For the remainder of the article, we are going to focus on these pipelines intimately.
This part particulars methods to combination uncooked occasions, such because the demand and provide occasions, by their geospatial and temporal dimensions, in addition to by world product (UberX, and so on.) for any given hexagon (c.f., right here). The simplified computation algorithm is as follows:
- Rely the variety of uncooked occasions from distinct riders and drivers by hexagon and world product kind in a 1-minute window
- Apply the Kring Easy to a number of rings, as much as ring-20 (mentioned later) on the 1-minute window
- Combination the smoothed values of every ring on a number of sliding window sizes as much as 32 minutes
In whole, one real-time pipeline generates 54 options for a hexagon every minute, utilizing the mixture of 9 rings (0, 1, 2, 3, 4, 5, 10, 15, 20), and 6 window sizes (1, 2, 4, 8, 16, 32).
Subsequent, we focus on step 2 of the algorithm:
The Kring Easy course of calculates the geospatial aggregation by broadcasting the occasion counts of a hexagon to its Kring neighbours. In different phrases, the characteristic worth of a hexagon for a selected ring takes under consideration the occasion counts from all hexagons inside that ring.
With the intention to calculate the characteristic worth aggregated on the ring R for a given hexagon H, the equation is:
- Num(i) is the variety of hexagons of ring i
- Nij is the jth hexagon of ring i
- f(H, 0) is the variety of occasions originated from Hexagon H
So let’s examine the next instance to see methods to compute the values of three options: ring 0, ring 1, and ring 2 of the hexagon A, following the equation:
Num(0) = 1
Num(1) = 6
Num(2) = 12
f(A, 0) = 1
f(A, 1) = (f(A, 0) + f(B1, 0) + f(B2, 0)) / (Num(0) + Num(1)) = (1 + 2 + 1) / 7 = 4 / 7
f(A, 2) = (f(A, 0) + f(B1, 0) + f(B2, 0) + f(C1, 0) + f(C2, 0) + f(C3, 0)) / (Num(0) + Num(1) + Num(2)) = (1 + 2 + 1 + 3 + 2 + 1) / (1 + 6 + 12) = 10 / 19
The pipeline follows the equation to calculate the values of options for a number of ring sizes, as much as 20.
After the Kring Easy completes for a one-minute window, step 3 of the algorithm is to additional combination the smoothed occasion counts on bigger home windows, as much as 32 minutes. With the intention to calculate the aggregation on a bigger window for a given hexagon H, the equation is:
- T is the beginning timestamp of a window
- W is the window dimension in minutes
- q(H, T, 1) is the smoothed occasion rely from the Kring Easy
Determine 3 under demonstrates methods to compute the characteristic worth of hexagon A for a 2-minute window:
- The smoothed occasion counts from Kring Easy for home windows – W1 and W2 – are 1.0 and three.0 respectively, and are emitted at T0 + 1min and T0 + 2min, respectively
- the characteristic worth for the 2-minute window is 2.0 by following the equation above utilizing the smoothed occasion counts, which falls into the time vary of (T0, T0 + 2min)
This part makes use of the demand pipeline as the instance for instance methods to implement the characteristic computation algorithm in Apache Kafka and Apache Flink, and methods to tune the true -time pipeline.
Logical Jobs Topology
Determine 4 under illustrates the logical DAG of the streaming pipeline to calculate the demand options. For all of the home windows whose sizes are higher than 1 minute, they’re sliding home windows, and these home windows can be sliding by 1 minute, which implies an enter occasion might be included inside 63 home windows: 32 + 16 + 8 + 4 + 2 + 1.
The desk under lists the functionalities for main operators within the logical DAG:
|Kring Easy on 1-Min||This operator applies the Kring Easy algorithm.|
|2-Min, 4-Min, 8-Min, 16-Min, 32-Min||These operators combination for these home windows sliding by 1 minute, utilizing the smoothed calls for returned from Kring Easy.|
|Merge Home windows||This operator collects the aggregated outcomes from all upstream home windows (1, 2, 4, 8, 16, 32), then packs them right into a single file for persistence.
For instance, at 1:32am, the operator will emit a file together with the demand options for time home windows:
Desk 1: Logical Operators of Demand Pipeline
The Streaming Pipeline’s Knowledge Quantity
This part lists the information volumes for the demand pipeline:
- The typical enter price of Kafka matters: 120k/s
- Rely of Hexagons: 5M
- Rely of cities: 1500
- Common and Most counts of Hexagons per metropolis: 4K and 76K
- Common counts of demand occasions by Hexagon at 1 minute: 45
- Hexagon counts of ring-20: 1261
It’s clear that the pipeline has excessive quantity, intensive computation, and huge states to handle. The primary model was truly constructed as per the logical DAG, which couldn’t run stably (as proven within the dashboard under due), as a result of points together with backpressure and OOM. Contemplating that we’re concentrating on near-real-time latency (lower than 5 minutes), there’s a actual problem forward of us to construct a secure, working pipeline.
The right way to Optimize
This part discusses methods to tune this streaming pipeline. At Uber, we’ve developed a framework of efficiency tuning for streaming pipelines, in addition to an end-to-end integration check framework. Devoted integration checks are developed earlier than kicking off the precise tunings, permitting us to refactor or optimize a streaming pipeline with confidence that the pipeline will nonetheless generate right outcomes, much like how unit checks shield us from regression. These integration checks turn out to be extraordinarily useful over the entire technique of optimization.
Subsequent, we are going to introduce the efficiency tuning framework.
Efficiency Tuning Framework
As proven within the interior triangle of the determine under, our framework focuses on 3 areas: Community, CPU, and Reminiscence, measured and monitored with the metrics served by Uber’s uMonitor system. The vertices of the outer pentagon point out the most important domains that might be explored for optimization.
The desk under briefly explains the methods and potential impacts of every area:
|Area||Main Impacted Areas||Remarks|
|Controls the rely of containers/works for a streaming job.
|Controls how the messages needs to be grouped by key, and transferred between the upstream and downstream operators. Partition is likely one of the most necessary domains, and it impacts all areas because the messages should be se/der, which takes important share of CPU, and may set off extra runs of rubbish collector, because of the object creation at deserialization.
|Controls how an operator of a streaming pipeline interacts with exterior companies or sinks.
|Algorithm||CPU||It’s cheap to imagine that the affect of algorithms is best on the CPU.
Desk 2: The Domains of Efficiency Tuning
Subsequent, we focus on methods to optimize the pipeline.
We’ve utilized many optimizations onto the streaming pipeline, and a few optimization methods affect on a number of areas as described above. One specific method, Custom-made Sliding Window, has a big affect on all 3 areas, so we’ve a devoted part to debate it, in addition to one for storage.
The most important optimization methods are listed within the desk under:
|Method||Space / Area||Explanations|
|Fields exclusion||Algorithm||Uncooked Kafka messages have many fields that aren’t utilized by the streaming pipeline; therefore we’ve a mapper to filter out unused fields on the very starting of Job DAG.|
|Key encoding / decoding||Algorithm||International product kind UUID is used as a part of the output key. We encode the UUID (128 bits) with a byte (8 bits) through an inside encoding, after which convert again to UUID: writing the output to sink, which reduces the dimensions for each the reminiscence and the community payload.|
|Dedup||Algorithm||We’ve launched a 1-Min window to solely maintain one demand occasion per distinct rider earlier than the Kring Easy. The dedup, carried out with a cut back perform, has considerably lowered the message price to be 8k/s for the costly Kring Easy calculation. The trade-off is that the dedup window has launched yet another minute to the latency, which is mitigated with different methods.|
|Discipline Kind Choice||Algorithm||We’ve refactored the algorithm in order that we may select Integer as the information kind for the intermediate computation worth fairly than Double, which additional reduces the message dimension from 451 bytes to 237 bytes.|
|Merge Window Output||Distant Name||We may have 2 choices when writing the output into sink:
We selected the latter, as the previous choice may solely output as much as 2M/s to the sink, which merely couldn’t deal with this ingestion price.
Desk 3: Strategies of Community Optimization
As detailed above, the important thing enchancment was to have each fewer and smaller messages.
The methods for reminiscence are listed within the desk under:
|Method||Space / Area||Explanations|
|Object Reuse||Rubbish Collector||By default, objects will not be reused in Flink when passing messages between the upstream and downstream operators. We’ve enabled the article reuse when attainable to keep away from message cloning.|
|Enhance Containers||Partition||Because of the giant information quantity and states (every container may devour round 6G reminiscence) we’ve used 128 containers, every with one vcore, for the streaming pipeline.|
|Key encoding / decoding||Algorithm||International product kind UUID is used as the important thing of the output. We encode the UUID (128 bits) with a byte (8 bits) through an inside encoding, and convert again to UUID earlier than writing the output to Sink, which reduces the reminiscence dimension.|
|Fields projection||Algorithm||Uncooked Kafka messages have many fields that aren’t utilized by the streaming pipeline, therefore we’ve a mapper to filter out unused fields on the entry level of Job DAG, lowering messages’ reminiscence load on computation.|
|Discipline Kind Choice||Algorithm||We’ve refactored the algorithm to decide on Integer as the information kind for the intermediate computation worth fairly than Double, lowering the message dimension from 451 to 237 bytes.|
Desk 4: Strategies of Reminiscence Optimization
Be aware: some methods have additionally been included for the optimization on the community.
The methods utilized for CPU optimization are listed under:
|Method||Space / Area||Explanations|
|Message in Tuple||CPU||Flink gives the Tuple kind, which is extra environment friendly in comparison with POJO at serialization, because of the direct entry with out reflection. We’ve chosen Tuple for messages being handed between operators.|
|Keep away from boxing / unboxing||CPU||The streaming pipeline must name an inside library (H3) to retrieve the neighbours for a given hexagon. The API returns an array of Lengthy, resulting in pointless boxing/unboxing together with the computation. We’ve added one other API to return an array of the primitive kind as a substitute.|
|Cache for API||Distant Name||We’ve enabled an in-memory cache to enhance efficiency when changing the worldwide product kind, because the return from distant API doesn’t change that a lot.|
|Hexagon Index Kind||Algorithm||We transformed the default hexagon information kind from String to Lengthy, which has lowered the window aggregation perform’s time by 50%.|
|Kring Easy||Algorithm||We’ve re-implemented the Kring Easy algorithm to make it extra environment friendly.|
Desk 5: Strategies of CPU Optimization
Custom-made Sliding Window
The pipeline nonetheless couldn’t run easily with simply the tunings above, as a result of it must combination on a number of sliding home windows (2, 4, 8, 16, 32). The window aggregation has the next overheads, because of the have to partition the occasions by a key:
- De/Ser when passing messages from the upstream to window operators
- Message Switch over community
- Object being created at deserialization
- State administration and metadata required by window administration, such because the window set off
These overheads have added important strain to Rubbish Collector, CPU and Community. To make issues worse, the sliding window requires extra states in comparison with a tumbling or fixed-size window, as a result of one occasion must be saved in a collection of slided home windows. Take a 4-minute sliding window for example: given an occasion occurred at 2021-01-01T01:15:01Z, this occasion can be saved within the following 4-minutes home windows,
- 2021-01-01T01:12:00Z ~ 2021-01-01T01:16:00Z
- 2021-01-01T01:13:00Z ~ 2021-01-01T01:17:00Z
- 2021-01-01T01:14:00Z ~ 2021-01-01T01:18:00Z
- 2021-01-01T01:15:00Z ~ 2021-01-01T01:19:00Z
Because of the fan-out impact from the sliding window, the pipeline is below a lot of strain from state administration. To resolve these points, we’ve manually carried out the sliding window logic with an operator of FlatMap, with the next options:
- With Object Reuse being enabled, the occasions from the upstream operators are handed and reused, which avoids the partition and associated prices
- States are managed in reminiscence, so in impact every occasion solely has one copy of information
We’ve the next estimation with respect to the utmost required reminiscence to maintain the states in reminiscence:
Complete Reminiscence = Rely(Hexagon) * Rely(Product) * Max(window dimension) * sizeof(occasion)
= 3M * 6 * 32 * 237b
With the parallelism of 128, the reminiscence per container is round 1G, which is manageable. In manufacturing, the precise reminiscence is way under the utmost, as a result of not all hexagons have occasions for a time vary.
The effectivity from this personalized sliding window is exceptional, so we’ve efficiently re-used this operator for greater than 5 completely different use instances that required aggregations on a number of giant sliding home windows.
Last Job DAG After Tuning
After optimization, we find yourself with an easier job DAG, the place the personalized sliding window has changed the bigger window operators.
The pipeline has been working reliably, as proven within the following 24-hour dashboards:
Container Reminiscence Monitor:
To make it simpler for us to keep up the pipelines and re-use the sinks, we’ve additional refactored the pipeline DAG by separating the sink operator right into a devoted writer job in Flink, and connecting the computation and writer jobs with Kafka. This part focuses on the small print of this writer job.
For the mannequin being served, it can lookup the demand and provide info based mostly on the geo, time and product. We chosen Docstore (Uber’s in-house KV retailer resolution) as our storage.
We began with a docstore cluster which is shared by many use instances.
Listed below are the outcomes for inserting a one row-per-API name. Write QPS peaks round 13k, however more often than not it’s on the order of lots of.
We tried to write down these rows in batches to see whether or not it could enhance throughput. To extend the batching effectivity, we partitioned the information based mostly on the shard quantity within the Docstore. Nevertheless, write QPS is decrease after batching is utilized. After we dug deeper, we discovered it was because of the cardinality of a dimension of metrics emitted within the streaming job is just too giant. We modify that dimension to a continuing string as a substitute of a random UUID. The write QPS can attain about 16k.
Earlier than writing to Docstore, we first write the information to a Kafka subject. After disabling the Kafka sink, we are able to see round a ten% enhance for write QPS.
Write QPS doubled to 34k after we modified the per-shard batch dimension to 50. We’ve additionally tried batch dimension 100 and 200. For batch dimension 100, write QPS will increase to 37k (about 20% enhance).
After altering batch dimension to 200, not a lot distinction (c.1k) was noticed.
Within the following desk, we checklist QPS below completely different configurations:
|Fixing the metrics cardinality downside||16k|
|After disabling writing to kafka subject||17.6k|
|Batch dimension 50||34k|
|Batch dimension 100||37k|
|Batch dimension 200||38k|
Desk 6: Throughput below completely different batch sizes
Flink job parallelism is one other parameter we tuned to enhance the QPS.
After updating the parallelism of the writer job to 256, the write QPS was round 75k, greater than doubled. Batch dimension is 200. With parallelism 1024, we see the QPS reaches 112k. Nevertheless, we see plenty of timeout errors already. After altering batch to 50, write QPS is round 120k.
|Job Parallelism||Batch Dimension||Write QPS|
Desk 7: Throughput below completely different job parallelisms
For every Flink job, we’ve additionally tried utilizing a thread pool to extend the write QPS, with the next outcomes:
|Thread Pool Dimension||Job Parallelism||Batch Dimension||Write QPS|
Desk 8: Throughput below completely different thread pool sizes
If we use thread pool dimension 16, peak QPS is round 120k, however it isn’t very secure.
After we tried each optimization we may consider within the shared cluster, it nonetheless couldn’t attain the write QPS we wished. We requested for a devoted cluster to check.
We eliminated the Docstore sink and simply saved the FlatMap. If we eliminated the decision to the partitioner, 64 containers can deal with over 200k enter message price with out lagging.
We added the customized partition technique earlier than the FlatMap.
With 384 containers, Lagging was round 12 min. Partitioner latency varies from 0.2ms to 5ms. Growing to 512 containers introduced the lagging down to three min. Later we discovered that 0.2ms per partitioner name is the bottleneck. We added an area partitioner name cache to flatmap. The cache hits have been much like enter message price after 20 min.
Nevertheless, lagging saved growing:
Again-pressure is on the customized partition stage.
Updating parallelism to 128 successfully eliminated any lag from the pipeline. Every DC can write 300K QPS with none downside.
We tried 3 completely different schema to see the information dimension distinction. The primary makes use of one column for every (ring dimension, time bucket, provide/demand) tuple. The second makes use of one map for demand and one map for provide. The third one teams 7 hexagons at granular stage 9 into one row.
With 6 days of information, we get the information sizes like this:
|No compression||With compression|
Desk 9: Compression below completely different information schemas
After enabling the compression, we’re seeing round 60% disk financial savings on all 3 tables.
Throughout testing, we discovered some latency points: P99 latency is round 150ms. It’s unacceptable for our pricing workflow. By way of debugging we discovered that every partition key has many rows–round 6k. This implies our database engine must scan at the least 6k rows after which will apply the filtering handed within the Question. Because the partition key’s dimension grows, it causes periodic spikes of 200msec or so. However we realized that TTL can also be set for this desk, so what we’ve performed is deployed a scorching patch in Question to limit the consequence to solely rows that aren’t expired, after which apply the filtering handed within the question. This lowered the scan on the underlying engine, and P99 latency dropped to 10ms.
Powering machine studying fashions with close to real-time options will be fairly difficult, as a result of computation logic complexity, write throughput, serving SLA, and so on. On this weblog, we launched a few of the issues that we confronted and our options to them, within the hope of aiding our friends in related use instances.
If you have an interest in becoming a member of the Market Intelligence staff, tackling challenges for giant scale streaming (e.g. Flink, Apache Samza), OLAP (e.g. Elasticsearch®) and ML (e.g. ETD prediction), please apply to affix our staff!
Apache®, Apache Cassandra®, Apache Flink, Apache Kafka, Apache Pinot, Apache Samza, Cassandra®, Flink, Samza, and Kafka are both registered emblems or emblems of the Apache Software program Basis in the US and/or different nations. No endorsement by The Apache Software program Basis is implied by way of these marks. Elasticsearch is a registered trademark of Elasticsearch BV.