Constructing Scalable Streaming Pipelines for Close to Actual-Time Options

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.


Determine 1: Simplified Overview of the Structure



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:

  1. Rely the variety of uncooked occasions from distinct riders and drivers by hexagon and world product kind in a 1-minute window
  2. Apply the Kring Easy to a number of rings, as much as ring-20 (mentioned later) on the 1-minute window
  3. 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:


Kring Easy

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:





The place:

  • 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


Determine 2: Ring 0, 1, 2 of Hexagon A


The pipeline follows the equation to calculate the values of options for a number of ring sizes, as much as 20. 


Temporal Aggregation

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:




The place:

  • 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)


Determine 3: Aggregation of a 2-minute Window for Hexagon A



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.

Determine 4: Logical DAG of Demand Pipeline


The desk under lists the functionalities for main operators within the logical DAG:

Operator Functionalities
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:

  • 1-min (1:31 – 1:32)
  • 2-min (1:30 – 1:32)
  • 4-min (1:28 – 1:32)
  • 8-min (1:24 – 1:32)
  • 16-min (1:16 – 1:32)
  • 32-min (1:00 – 1:32)

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.


Reminiscence Monitor:

Determine 5: Dashboard of the Used Reminiscence


Lagging Monitor:

Determine 6: Dashboard of the Lagging



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.

Determine 7: the Framework of Efficiency Tuning


The desk under briefly explains the methods and potential impacts of every area:

Area Main Impacted Areas Remarks
Parallelism CPU


Controls the rely of containers/works for a streaming job.

Use Circumstances:

  • If a job’s enter price is 16k/s, and every container may course of 2k/s at most, then the parallelism needs to be at the least 8 or 16k / 2k; for a similar job, if the Kafka subject has fewer than 8 partitions, then it is smart to bump the subject’s partitions.
Partition Community



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.

Use Circumstances:

  • Be cautious with the skewed partition, which may create hotspots, slowdown, or trigger lagging for a streaming pipeline
Distant Name Community


Controls how an operator of a streaming pipeline interacts with exterior companies or sinks.

Use Circumstances:

  • If the distant name is the bottleneck, to think about including Redis cache or in-memory cache
  • Redis cache additionally entails distant name, which may turn out to be a bottleneck when retrieving a cached merchandise with giant information dimension
  • Take into account with async
Algorithm CPU It’s cheap to imagine that the affect of algorithms is best on the CPU.

Use Circumstances:

  • Take into account sampling the occasions if the pipeline’s window has giant quantity of them, reminiscent of to pattern occasions for the calculation of percentiles
Rubbish Collector Reminiscence


Use Circumstances:

  • Reuse the message or object when passing messages between the upstream and downstream operators if attainable, to keep away from pointless object creation/elimination
  • Be cautious with pointless boxing/unboxing, which may add up when processing excessive quantity information
  • Have pipeline-specific reminiscence administration, utilizing pre-allocated reminiscence

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.


Community Optimization

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:

  • Let home windows of all supported dimension (1-32 Min) to immediately write into sink
  • Introduce an additional window to merge these home windows’ output right into a single file first, then write 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.


Reminiscence Optimization

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.


CPU Optimization

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

                       = 136G

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

Determine 8: Last DAG of Demand Pipeline


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:


Lagging Monitor:

Determine 9: Dashboard of Displaying the Lagging After Optimization


Container Reminiscence Monitor:

Determine 10: Dashboard of Displaying Reminiscence Utilization After Optimization


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. 

Determine 11: Write QPS will not be secure with one row per API name




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: 

Write QPS 
First model 150
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
256 200 75k
1024 200 112k
1024 50 120k

Desk 7: Throughput below completely different job parallelisms


Thread Pool

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
32 128 50 64k
128 128 50 62k
128 256 50 80k
16 256 50 120k

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.  


Partitioner Tuning

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:

Determine 12: Job lagging retains growing


Again-pressure is on the customized partition stage.

Determine 13: Topology of the job and backpressure 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.


Knowledge Dimension

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
driver_pricing 96.3TB 36.8TB
driver_pricing_map 105.6TB 43.6TB
driver_pricing_map_group 328TB 132.6TB

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. 

Determine 14: Serving latency drops from 150ms to 10ms after the optimization



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.

Source link



Please enter your comment!
Please enter your name here