Lyncredible Navigating the tech stack of engineering and management

Calling dibs on DIBS

Oct 30th, 2023

I joined Uber in October 2015 to work on the brand-new Driver Incentives Backend System. Jonah Cohen, my soon-to-be manager, called dibs on naming it DIBS just a few weeks before I arrived. Fast forward eight years, and now I am calling dibs on recounting the 1 I drew inspiration from Matt Basta’s captivating piece titled No sacred masterpieces, which recounts how he “built Excel for Uber and they ditched it like a week after launch”. Intriguingly, both tales unfolded during the same time frame centered around the same driver incentives program. Yet, surprisingly, even after spending over 5 years together at Stripe following our Uber stints, Basta and I have never crossed paths..

 

Driver Incentives and Carrots

2015 was the heyday of price wars in the ride-sharing market. Most riders likely encountered aggressive promotions and jaw-droppingly low prices, clearly underwritten by the platforms themselves. However, the rider subsidies were mere shadows compared to the incentives 2 To grasp the scale of the driver subsidies, consider this: drivers often earned more from incentives than they did from ride payouts. upon drivers. At first, these incentives served to lure drivers, ensuring ample supply aligned with anticipated high demand. But as competition intensified, such incentives evolved from a recruitment tool to an essential strategy for 3 I provided a theory on the intense price war in a previous post..

Uber operated by cities, with driver incentives locally managed by each city’s 4 In many ways, Uber’s early operations echoed the franchise model of McDonald’s. City managers acted as the franchise owners, while the Uber app, internal tools, and playbooks served as the secret recipes for whipping up Big Macs. Many innovations sprouted from these city operations team, only to be replicated and scaled later with the help of the central teams at the headquarters.. These incentives were structured as weekly campaigns for drivers. A typical campaign might state: earn an additional $1,000 next week if you complete a total of 40 trips either from or to downtown Manhattan, between 7am and 9am or between 4pm and 7pm, and from Monday to Friday.

Like any other great business software, driver incentives at Uber started as spreadsheets. Some tenacious Driver Operations people, or DOps, started the program by emailing campaign details to drivers in their cities at the beginning of each week. When the week ended, they wrote 5 If ChatGPT had been available back then, the DOps would surely have used it to write SQL queries. based on the campaign rules to count qualifying drivers and calculate earnings. The incentives were paid out by uploading CSV files with driver UUIDs and dollar amounts.

Sometime in 2014, the Driver Incentives product team was formed at Uber’s headquarters in San Francisco. They built out the software to manage the campaigns from creation to payout. Named 6 In later years, given the broadly critical stance of the media towards Uber, the name Carrots might have sparked a minor PR hiccup. Fortunately, it didn’t. The team simply saw it as an easily recognizable codename., the software was a game-changer to DOps who routinely worked 12+ hours a day and 6+ days a week. Here’s the 7 The following marketing content was mine, but the thrill was ChatGPT.:

 

From Incentica to DIBS

There was just one hiccup: Carrots used the same 8 Vertica, for those unfamiliar, is an analytics database that’s designed for very fast queries over very large sets of mostly read-only data. database that DOps ran their SQL queries on. The team bought the beefiest Vertica box on the market, affectionately dubbing it Incentica. However, even this powerhouse was nearing its breaking point. Vertica wasn’t built for horizontal scaling. It could not house all of Uber’s data or deal with the erratic, fast-growing query demands.

Now, here’s a quirky thing: Uber’s idea of a week ended at 4am every Monday, based on each city’s local time. This meant that, for the most part, Incentica had a leisurely pace throughout the week. However, come Monday morning in East Asia, queries started flooding in. And as the clock struck 4am in city after city, the floodgates opened wider. The real kicker? The Carrots team had to constantly babysit these queries. So, their workweek effectively began at noon on Sundays to align with 4am Monday in Beijing. And if you were the unlucky on-call engineer, you were probably pulling an all-nighter on Sunday to ensure Incentica could withstand the onslaught from the Middle East, Europe, and eventually the Americas.

Then there was the ETL pipeline responsible for transferring data from the primary trips database to Incentica. The pipeline was delayed and lossy due to legacy constraints, which meant the calculations could still end up off the mark despite all the waiting and babysitting.

Hailing from Seattle, the DIBS team was tasked with rectifying these issues. Our solution seemed straightforward. DIBS would harness streaming-based aggregation to update calculations for each driver and every campaign in real time. Picture this: hundreds of worker nodes running Kafka consumers, all eagerly subscribed to trip completion events. The workers would coordinate to partition and process the events, making the system 9 The system was horizontally scalable up to the number of virtual partitions in Kafka, which was 4,096 at the time.. The streaming architecture not only distributed the workload evenly throughout the week, eradicating those dreaded Monday query spikes, but it also had a cherry on top: drivers could monitor their real-time progress in the Uber app as the week unfolded. As an added safety net, We opted to shield against 10 We also contemplated using Bloom filters in the design to quickly detect duplicates, but they were never implemented. Even for the most industrious driver, there just weren’t enough trips to justify it. by recording all processed trip UUIDs for each driver and campaign. Looking back, the design might seem deceptively simple or even naive, but guess what? It did the trick.

We called dibs on the first full week of 2016 for DIBS’ prime-time debut.

Scaling to sound sleep

And that was the week when I paid my fair share of Sunday all-nighters.

DIBS actually operated smoothly throughout Sunday night. Yet, both I and Greg, the Carrots on-call engineer, kept getting questions from DOps across various cities. They lamented the absence of calculation results for their campaigns. A closer look revealed that these campaigns had remained in the draft state all week. DOps were supposed to click the “launch” button before the week started to kick off streaming aggregation in DIBS. Their oversight was understandable; before DIBS, they had grown accustomed to waiting until week’s end to click “launch”. This was due to the batch-oriented nature of the Incentica solution, which could only process results after the week concluded.

That night I learned more about Kafka than the previous few months combined. Greg and I scrambled to “launch” the campaigns and then rewind the Kafka consumer offset to the previous Sunday. To expedite things, we added 11 We probably scaled out the number of Kafka consumer workers by 10x, although my memory is blurred. This was feasible because, first, the trips data was already spread across 4,096 virtual partitions, and second, Uber’s compute infrastructure was adept at elastically scaling containerized workloads. The first point allowed us to scale out horizontally as long the number of worker nodes did not exceed 4,096. The second point was even more impressive considering that Will Larson’s team built the compute infrastructure in Uber’s own physical data centers. to the Kafka consumer group, enabling DIBS to churn through the entire week’s trips in just a couple of hours. Ultimately, we managed to complete calculations for most campaigns just a few hours past the Monday 4am cut-off. The only challenge left? Tackling those sleep-depriving Sunday nights.

Going back in time

In the week that followed, we worked with DOps to launch campaigns on time. Every DOps individual we spoke to was understanding and sympathetic. They acknowledged the advantages of punctual campaign launches, such as allowing drivers to view their real-time progress towards campaigns within the app. While they wanted to help the engineers stop dreading about their on-call shifts, they also politely requested a buffer. What if a campaign was launched a few minutes late? Or if a mistake in the campaign rules was discovered later in the week?

That turned out to be a fascinating engineering problem for the DIBS team to solve. We built an automated backfill solution for late campaigns. To achieve this, we maintained two kafka consumer groups, both processing the same trips dataset. The primary one always processed trips in real time, while the secondary one was responsible for 12 Little did I know back then, but it turned out to be an anti-pattern to backfill by rewinding streaming consumer offsets. The proven way to do this was called a Lambda Architecture, where backfills ought to be performed using batch data processing techniques like Apache Spark. It did not matter though, because Uber had yet to build its lossless ETL pipeline to send trip data into its Hadoop-based Big Data Platform.. When a campaign was launched late, the backfill consumer group would automatically rewind its offset to match the campaign’s intended start time, and backfill any trips that had happened prior to the late-launch. We also introduced various optimizations, such as batching multiple late campaigns for backfilling and meticulously filtering out irrelevant trips, ensuring the backfill process was swift and efficient.

The automated backfill was a smashing success. Every Carrots campaign, even those with delayed launches, was promptly calculated and paid out with little human intervention. Both Carrots and DIBS engineers enjoyed 8 hours of sound sleep during their on-call shifts.

At least that was the case until DIBS launched in China two months later, when every assumption we made about the system had to be changed.

Everything was bigger in China

DIBS made its debut in China around March 2016, right when the rivalry between DiDi and Uber was escalating from fierce to downright incendiary. With a dense population and aggressive promotions/incentives, Uber was clocking millions of trips every week in each of China’s major cities. This volume dwarfed even the busiest Uber markets elsewhere in the world. At the peak, Shanghai saw ten times more trips than New York City in a typical week. The sheer scale would have completely destroyed Incentica. Fortunately, for DIBS, accommodating the colossal Uber China traffic was just a matter of deploying more worker nodes.

The real challenges was, interestingly and somewhat expectedly, in backfills. Uber’s DOps in China adopted a strategy of intentionally delaying campaign launches in Carrots until the week’s end. This tactic was devised to keep DiDi in the dark about Uber’s campaign rules, 13 Basta alluded to the same competition dynamics in his article, where Excel formulas to calculate incentives must be kept confidential because Uber was worried that DiDi sent spies to intern at Uber and steal such secrets.. Instead, they assured drivers that Uber would either match DiDi’s incentives or provide superior ones, but the precise formula would remain unknown until the week concluded.

This approach wreaked havoc on DIBS. Our primary, real-time consumer sit there all day doing nothing, while the secondary, back-fill consumer bore the brunt of the workload, meticulously processing every single trip for each campaign in the few hours following each week’s close. The surge in compute and network demand during backfills was staggering, scaling to levels about 1,000 times higher than the usual rate.

So we set out to scale DIBS for China. The key insight was that most changes to campaign rules had no impact on the partial aggregation logic. For example, an incentive of $1,000 demanding 100 weekly trips to qualify would utilize the same trip counter as another incentive of $2,000 requiring 180 weekly trips. This meant DIBS could do real-time aggregation of the partial result, like the trip count, and reuse them for varying campaign rules, as long as the incentive structure remained consistent.

Each week we launched 14 The dummy campaigns were not visible in the Uber Driver app. featuring common incentive structures, performed partial streaming aggregations throughout the week, and when the week wrapped up, reused the partial aggregation results for the actual campaigns launched by DOps. We even automated the process of detecting prevailing incentive structures and initiating these dummy campaigns. At last, our pursuit of uninterrupted slumber was realized.

The trip that existed before it didn’t

A mysteriously missing trip broke our peaceful dream later that spring. A DOps reached out to report that a driver was unfairly deprived of their incentive. According to DIBS, they fell short by a single trip. That was baffling. Uber’s Kafka setup ensured “at-least-once” event delivery, implying DIBS should 15 As is mentioned earlier in this post, DIBS recorded trip UUIDs to protect against duplicate event deliveries to avoid over-counting.. If this driver was denied incentive due to a missed trip, how many other drivers might have been similarly shortchanged?

To unravel this, we manually initiated a backfill for the contentious campaign. Somewhat unexpectedly, the backfill detected the elusive trip and confirmed the accurate trip count for the concerned driver. Bolstered by this discovery, we triggered backfills for all campaigns from the previous week, comparing the results pre and post-backfill. The scrutiny unearthed a minuscule yet consistent pattern of omitted trips.

There was an undeniable glitch within the primary, real-time streaming consumer. It seemed as though certain trips were momentarily non-existent in real-time but subsequently found their way into the system, only to be captured during backfills. The pressing question was, why?

The database that was not Kafka

To answer that question, we delved deeper into the origins of our streaming data. As it turned out, DIBS was not pulling data from a Kafka topic. Instead, it was sourcing from a Schemaless table. Schemaless was Uber’s in-house online datastore, designed for vital business data like rider, driver and trip details. Built on top of sharded MySQL tables, Schemaless strictly adhered to append-only writes, which was tailor-made for stream processing tasks akin to DIBS. To further its compatibility, the Schemaless team ingeniously developed client-side stream-consumer libraries that mirrored Kafka’s API. It is no wonder that this whole blog post, up until this point, depicted DIBS as if we were using Kafka the whole time.

The missed trip broke that abstraction though. To understand why, we need to first dive into the data schema in Schemaless.

The Schemaless schema

All Schemaless datasets were stored in MySQL with a physical schema that could be depicted 16 The physical schema was simplified to focus on the race condition. Check out Schemaless on Uber’s Engineering Blog for the complete definition.:

Column Name Column Type Note
ID BIGINT Primary key with auto-increment
UUID CHAR(36) Globally unique identifier, indexed
Created TIMESTAMP Timestamp auto-populated upon creation
Payload MEDIUMTEXT Serialized JSON payload

If we were to store a Uber trip object into a table like this, we would serialize the object into a JSON format, generate a UUID for the object, and append the resultant row to the table. The integer ID would be auto-generated by MySQL thanks to auto-increment. The Created timestamp will be auto-set with UTC_TIMESTAMP(). Here is a sample row in columnar format for clearer visualization:

Column Value
ID 42
UUID c160a108-91c5-4af4-8133-727c36f1e277
Created 2023-10-29T19:01:04.146Z
Payload {"city_id": "xxx", "amount_in_usd": 30, "distance_in_km": "18", ...}

Note this physical table was just one shard of many. Schemaless partitioned each dataset among 17 DIBS primarily consumed from the trips dataset in Schemaless, which had 4,096 shards.. The globally unique UUID functioned as the pivotal sharding key. The integer IDs were unique within each shard, but were expected to overlap across different shards.

The “Kafka” consumer

The “Kafka” consumer workers streamed data from Schemaless by taking advantage of the append-only nature of the underlying tables. Each shard was earmarked for 18 The converse was not true. Each worker could stream from multiple shards.. This worker ran an infinite loop to consistently poll for fresh rows to process. The loop’s mechanics looked like the following:

  1. Initialize per_shard_offset to 0.
  2. Run the SQL query: SELECT id, uuid, payload FROM charges WHERE id > $per_shard_offset to fish out new rows.
  3. Process the rows retrieved from Step 2 and update per_shard_offset to the highest ID identified among the new rows.
  4. Circle back to Step 2.

At first glance, the logic appeared impeccable. Since each shard was singularly managed by one specific worker, there wasn’t any room for race conditions between two different workers. So, with the mystery deepening, we documented our quandary and sought insights from the Schemaless team. Upon reviewing our notes, Rene Schmidt, the esteemed architect behind Schemaless, instantly pinpointed the issue, stating, “It’s a race condition due to Repeatable Read”.

The race condition

Imagine two new rows being appended to the table concurrently as the DIBS consumer ran its SELECT query. This happened all the time because many trips were completing around the world at any given time. It was not rare for two to end up in the same shard and to be written to the same MySQL table simultaneously. Suppose the last known highest ID was 42 just before these events. The table below outlines a possible sequence of events:

Time Rows Appender #1 Appender #2 DIBS consumer
t0 1-42 Idle Idle per_shard_offset = 42
t1 1-42 Begin Tx Idle Idle
t2 1-42 Append #43 Begin Tx Idle
t3 1-42 Idle Append #44 Idle
t4 1-42, 44 Idle Commit Tx SELECT ... WHERE id > 42 yields #44
t5 1-42, 43, 44 Commit Tx Idle Update per_shard_offset = 44

From the sequence, it’s evident that although Appender #1 initiated its transaction before Appender #2, the latter managed to append a row (with ID 44) and complete its transaction earlier. Unfortunately, the DIBS consumer executed the SELECT query post Appender #2’s commit but pre Appender #1’s commit, getting only row #44 in return. This means in its subsequent scans, DIBS would bypass any rows with IDs below than 44. As a result, row #43 slipped through the cracks, remaining undetected from the DIBS consumer’s point of view.

The workaround

The root of this race condition traced back to the default transaction isolation level of Repeatable Read in MySQL / InnoDB. It could be prevented by changing the transaction isolation level to Serializable. In that mode, the SELECT transaction would create a range lock on ID, essentially locking the range (42, +∞) in our earlier example. That lock would conflict with the single row locks established by appenders, such as a single row lock of 43 by Appender #1 in the same example. This means the SELECT transaction would wait for all in-progress append transactions to finish before executing the query, thus avoiding the race condition.

Yet, switching to Serializable mode was not free. It would incur a huge performance penalty, forcing many transactions to execute in serial rather than concurrently. The impact would reverberate any production systems interfacing with the database. Given the resulting dramatic plunge in throughput, this approach was not viable.

Our eventual solution was a bit of a workaround tailored to the Schemaless stream consumer library. The SELECT query was modified to exclude recent rows by examining the created timestamp:

SELECT id, uuid, payload
FROM trips
WHERE id > $per_shard_offset
  AND created < TIMESTAMP(
    DATE_SUB(UTC_TIMESTAMP(), INTERVAL 2 MINUTE))

By imposing a condition for the created timestamp to be at least 2 minutes old, the SELECT query effectively ignored all rows appended by recent transactions, and punted them to a subsequent loop iteration. The choice of a 2-minute buffer was enough to sidestep the race condition without adding too much delay. This adjustment avoided degrading overall throughput of the entire system, incurring only a minor delay on the stream consumer side.

DIBS until Teleportation

DIBS ultimately lived up to its promise of lossless aggregation. Incentica could finally be retired.

The entire DIBS team flew down to San Francisco to attend the joint Incentica-shutdown party. While in town, I had the chance to engage with several members of Uber’s China Growth organization, the largest user of DIBS. I was surprised to discover that numerous teams within China Growth had use cases mirroring Driver Incentives. Each program revolved around the theme of giving rewards to eligible riders or drivers upon meeting certain criteria.

On the following day, a 19 TK is the preferred moniker for Travis Kalanick, co-founder and then CEO at Uber. Q&A with China Growth was on the agenda, so I asked Travis a question: “It seems almost half of China Growth are doing various forms of promotion and incentive programs. Assuming there will come a day when we stop the aggressive spending, what innovations should we be focusing on now to differentiate our product when that time arrives?”

TK’s answer was memorable: “Some day Uber will replicate 20 See Transporter (Star Trek) on Wikipedia. and teleport people from Point A to Point B. But until then, we will keep doubling down on incentive programs because everyone else is spending like crazy.”

The Q&A was hosted within the confines of Uber’s headquarters at 1455 Market Street. Personally, its interior design reminded me strongly of the USS Enterprise. Beyond sci-fi analogies, it was evident that TK was alluding to the future of self-driving cars. I had always been 21 See my previous post Uber had no upside which covered my doubts about the potential impact of self-driving technology on Uber’s business. about the strategy of betting on self-driving technology while bleeding cash in the core business. TK’s response did little to allay those concerns.

While in SF, I also had the privilege to present a company-wide tech talk on DIBS. As expected, the automated backfill solution was a hot topic of discussion. It was unconventional, caused spiky traffic patterns, and strained both upstream and downstream systems. I recognized and validated the concerns, but also emphasized the short-term necessity of the solution as Uber was spending billions of dollars on driver incentives per year. The vision was to transition to a more robust solution once the 22 Sitting in the audience, Zheng Shao was one of the many who asked questions about the efficiency and scalability of the DIBS backfill. He would go on to build Uber’s Big Data Platform, making it possible for DIBS to migrate to a much more scalable Lambda Architecture. was operational. “Or perhaps, when Teleportation becomes a reality,” I mused privately.

DiDi called dibs on us

Neither was fast enough. On August 1st, 2016, less than one year after the start of the DIBS project, Uber sold its China operations to DiDi, receiving a minority stake in the latter as part of the deal.

Most of the sophisticated features we had developed for DIBS lost their relevance overnight. Within a month, ownership of the DIBS system was transferred to the Carrots team. Subsequently, every member of the DIBS team embarked on new ventures within Uber.

Throughout the DIBS project, there had been internal reservations regarding the overarching strategy of the price war. However, such concerns never impeded our dedication to crafting the best possible solution within our set parameters. With the change in direction, there was no need to dwell on past accomplishments. After all, business is business, or in Basta’s words, “No sacred masterpieces”.

 
Share via

Related Posts

Recent Posts

© 2020 - 2024 Yuan Liu All Rights Reserved.
Posts  |  Feed  |  Newsletter  |  Tech Stack  |  About