Gewährleistung der Skalierbarkeit eines Online-Taxidienstes am Beispiel von Uber

Abstract: Skalierbarkeit gehört zu den zentralen Themen verteilter Systeme. Mit zunehmender Nachfrage und regionaler Ausbreitung eines verteilten Dienstes sind im Lauf der Zeit erhebliche Einschränkungen in der Leistung und Zuverlässigkeit zu erwarten. Insbesondere bei einem Taxidienst, dessen Dienste weltweit über eine mobile Anwendung angeboten werden, müssen spezifische technische Gegebenheiten wie Echtzeitübertragung und die Häufigkeit von Nutzeranfragen berücksichtigt werden, um entsprechend angemessene Skalierungsmaßnahmen zu treffen. Für die Gewährleistung einer optimalen Leistung und Verfügbarkeit unabhängig von der Nachfrage durch Endnutzer ist im Wesentlichen zu erkennen, dass eine klare Einteilung von Rollen, Verantwortlichkeiten und Ressourcen der verfügbaren Hardware bezüglich geographischer und anwendungsinterner Zuständigkeiten unerlässlich ist. Für die Erforschung passender Skalierungstechniken werden verschiedene Techniken zur Skalierung eines verteilten Online-Taxivermittlungsdienstes am Beispiel des Taxidienstes Uber erläutert und diskutiert.

Index Terms: distributed systems, scalability, software architecture, sharding, partitioning, microservices

I. Motivation

Skalierbarkeit zählt zu den zentralen Herausforderungen verteilter Systeme. Unternehmen, die verteilte Anwendungen kommerziell betreiben, besitzen ein wirtschaftliches Interesse, ihre Dienste in mehreren Regionen der Welt bereitzustellen, um eine möglichst große Nutzerbasis zu erreichen. Eine globale Verbreitung geht jedoch mit erheblichen technischen und wirtschaftlichen Herausforderungen einher. Insbesondere muss sichergestellt werden, dass die Anwendung auch bei einer hohen Anzahl gleichzeitiger Anfragen sowie einer großen Zahl beteiligter Server weiterhin eine geringe Antwortzeit aufweist.

Im Kontext verteilter Systeme beschreibt Skalierbarkeit die Fähigkeit einer Anwendung, mit zunehmender Last effizient umzugehen und ihre Leistungsfähigkeit durch den Einsatz zusätzlicher Ressourcen aufrechtzuerhalten. Die architektonische Ausgestaltung eines skalierbaren Systems zielt darauf ab, auch bei einer stark steigenden Anzahl von Nutzeranfragen und einer weltweit verteilten Nutzerschaft kurze Antwortzeiten zu gewährleisten. Dabei soll verhindert werden, dass die Menge der Hardware-Ressourcen zu einem Engpass für die Leistung und Zuverlässigkeit des Systems werden.

II. Grundlagen

In diesem Kapitel werden einige zentrale Aspekte von Skalierbarkeit zusammenfassend erklärt, die die Grundlage für das Verständnis spezifischer Konzepte bilden.

A. Horizontale und vertikale Skalierung

Ein zentrales Konzept der Skalierbarkeit ist die Unterscheidung zwischen horizontaler und vertikaler Skalierung. Diese beiden Ansätze beschreiben unterschiedliche Methoden zur Steigerung der Leistungsfähigkeit einer verteilten Anwendung.

1) Horizontale Skalierung

Unter horizontaler Skalierung wird die Erweiterung der verfügbaren Hardware-Ressourcen durch die Erhöhung der Anzahl eingesetzter Rechner verstanden. Die Leistungsfähigkeit einer Anwendung wird dabei gesteigert, indem zusätzliche Server in das System integriert werden. Dies ermöglicht nicht nur die Verteilung von Nutzeranfragen auf mehrere Server und damit die Reduzierung der Auslastung einzelner Systeme, sondern auch eine Reduktion der Latenz durch geographische Verteilung von Serverstandorten. Horizontale Skalierung bringt jedoch die Anforderung mit sich, die verteilten Systeme angemessen zu koordinieren, sodass in der Verarbeitung der Anfragen und dem Stand der Daten keine Inkonsistenzen entstehen. Dies umfasst beispielsweise die Weiterleitung von Anfragen an geeignete Serverinstanzen. In diesem Zusammenhang spielt das Konzept der Location Transparency eine wichtige Rolle: Die Verteilung der Anwendung soll für die Nutzer transparent bleiben, sodass das System trotz seiner verteilten Architektur als eine einzelne Anwendung wahrgenommen wird. [1]

2) Vertikale Skalierung

Vertikale Skalierung bezeichnet die Leistungssteigerung eines einzelnen Rechners durch die Verbesserung seiner Hardwareausstattung. Eine Anwendung kann vertikal skaliert werden, indem leistungsfähigere Komponenten eingesetzt werden, wodurch die physischen Kapazitäten des Host-Systems erhöht werden. Dies kann verschiedene Hardware-Komponenten betreffen, darunter den Prozessor, den Arbeitsspeicher, sowie den persistenten Speicher. [1]

Im Vergleich zur vertikalen Skalierung bietet die horizontale Skalierung in der Regel größere Erweiterungsmöglichkeiten. Die vertikale Skalierung ist durch die Leistungsgrenzen der verfügbaren Hardware beschränkt, da die Aufrüstung eines Systems nur solange möglich ist, wie leistungsfähigere Komponenten zur Verfügung stehen. Mit zunehmender Hardware-Leistung wird schließlich ein Punkt erreicht, an dem keine weitere signifikante Verbesserung durch den Austausch einzelner Komponenten möglich ist.

Im weiteren Verlauf wird der Schwerpunkt auf der horizontalen Skalierung liegen. Die vertikale Skalierung ist zwar ebenfalls von Bedeutung und sollte nicht als irrelevant betrachtet werden, jedoch bieten horizontale Skalierungsansätze in vielen Szenarien größere Flexibilität und Erweiterungsmöglichkeiten. Auf die vertikale Skalierung wird dennoch zu einem späteren Zeitpunkt detaillierter eingegangen.

B. Replizierung

Replizierung bezeichnet die mehrfache Bereitstellung einer Anwendungskomponente auf unterschiedlichen Servern. Dabei kann sich die Replizierung sowohl auf Daten, beispielsweise Dateien eines File-Hosting-Dienstes wie Google Drive, als auch auf die Anwendung selbst beziehen, indem ein Dienst parallel auf mehreren Servern ausgeführt wird. [1]

Daten werden repliziert, indem identische Datenbestände auf mehreren Speichermedien gleichzeitig vorgehalten werden. Fällt eines dieser Speichermedien aufgrund eines Defekts aus, bleiben die Daten weiterhin verfügbar, da auf ein anderes Replika zurückgegriffen werden kann. Darüber hinaus dient die Replizierung dazu, Daten auf mehrere geographische Standorte zu verteilen und dadurch den Zugriff lokaler Server auf benötigte Daten zu beschleunigen. Dieses Prinzip lässt sich nicht nur auf Daten, sondern ebenso auf Anwendungskomponenten anwenden. [1]

Replizierung ist jedoch auch mit Herausforderungen verbunden. Insbesondere muss die Konsistenz zwischen den replizierten Daten gewährleistet werden. Dies erfordert zusätzliche Anwendungslogik und verursacht einen entsprechenden Performance-Overhead. Zusätzlich kann die für die Konsistenzsicherung verantwortliche Logik selbst zu einem Engpass in der Skalierung werden, da die zugrunde liegenden Algorithmen je nach Implementierung potentiell nur begrenzt skalierbar sind. Während des Replizierungsprozesses muss sichergestellt werden, dass alle Replikas konsistent aktualisiert werden. Zudem ist festzulegen, wie mit Problemen in der Concurrency umgegangen wird, insbesondere in Situationen, in denen Änderungen an einem Replika vorgenommen werden, bevor frühere Aktualisierungen auf allen anderen Replikas wirksam geworden sind. [1]

C. CAP-Theorem im Kontext von Uber

Das CAP-Theorem besagt, dass in einem verteilten System die drei Eigenschaften Consistency, Availability und Partition Tolerance nicht gleichzeitig vollständig gewährleistet werden können. Im Falle einer Netzwerkpartition kann ein System höchstens zwei dieser Eigenschaften gleichzeitig erfüllen. [1]

Die Architektur von Uber priorisiert grundsätzlich Verfügbarkeit und Partitionstoleranz. Weshalb dabei keine vollständige Konsistenz gewährleistet werden kann, wird in Kapitel V-E bezüglich der Event-basierten Architektur näher erläutert. Im Zusammenhang mit geographischem Sharding (wie in Kapitel V-B beschrieben) wird zudem ersichtlich, dass eine globale Konsistenz über verschiedene Regionen beziehungsweise Shards hinweg nicht zwingend erforderlich ist, da die Anwendung überwiegend innerhalb der jeweiligen geographischen Region relevant ist. Die Gewährleistung von Konsistenz innerhalb eines einzelnen Shards stellt hingegen in der Regel kein wesentliches Problem dar, da die geographische Nähe zwischen den beteiligten Systemen typischerweise zu geringen Latenzen führt.

Für sämtliche Komponenten einer Anwendung muss nicht zwangsläufig dieselbe Ausprägung des CAP-Theorems vorliegen. So kann die Datenbankschicht Konsistenz und Partitionstoleranz priorisieren, während die ereignisgetriebene Architektur auf Verfügbarkeit und Partitionstoleranz ausgelegt ist.

III. Anforderungen an die Funktion eines verteilten Taxidienstes

Im Folgenden werden Aspekte genannt, die zum Einen als allgemeine Anforderungen an die Architektur eines verteilten Taxidienstes verstanden werden können, und gleichzeitig den Grund für die Notwendigkeit von Skalierung darstellen.

Konsistenz: Nutzerdaten müssen schlüssig bleiben, selbst wenn der Nutzer in eine andere Region der Welt wechselt. [2]

Aktualität der Daten: Neu erzeugte Datensätze müssen innerhalb von Sekunden vom Nutzer abrufbar sein. [2]

Verfügbarkeit: 99.99% Verfügbarkeit muss in jeder geographischen Region, die vom Taxidienst abgedeckt wird, gewährleistet sein. [2]

Kosten: Als Firma mit niedrigen Margen ist es insbesondere wichtig, Kosten zu sparen. [2]

Latenz: Requests sollen höchstens wenige Sekunden brauchen, um vom Server empfangen und verarbeitet zu werden. Im Gegenzug sollen Server-Antworten genauso lange brauchen, um beim Nutzer anzukommen. [2]

Echtzeit: Uber muss Nutzern in Echtzeit Fahrer zuweisen. Somit ist Uber nicht nur eine verteilte Anwendung, sondern auch eine Echtzeitanwendung. [2]

Die erwähnten Anforderungen stellen indirekt zugleich die grundlegenden Ziele von Skalierung dar. Durch Skalierung werden Konsistenz, Aktualität, Verfügbarkeit, Kosten, Latenz, und Echtzeit von der Anzahl der Nutzeranfragen und dem geographischen Ort eines konkreten Nutzers voneinander entkoppelt. Insbesondere der geographische Aspekt wird in den folgenden Kapiteln näher erläutert.

IV. Übersicht über den architektonischen Wandel des Taxidienstes Uber

In den Anfangsjahren bot Uber einen deutlich weniger umfangreichen Dienst an. Zum Zeitpunkt der Unternehmensgründung im Jahr 2009 beschränkte sich das Angebot ausschließlich auf die Vermittlung von Taxifahrten und war zunächst auf die Stadt San Francisco begrenzt. [3]

Die ursprüngliche Uber-Anwendung war als monolithisches System konzipiert. Ein Monolith bezeichnet eine Softwareanwendung, die als eine einzige, zusammenhängende Einheit entwickelt, bereitgestellt und ausgeführt wird. Sämtliche funktionalen Teilbereiche der Anwendung sind innerhalb derselben Codebasis integriert und weisen eine enge Kopplung auf. Monolithische Anwendungen sind daher weitgehend in sich geschlossen und für die Ausführung ihrer Kernfunktionalitäten nicht auf externe Anwendungsdienste angewiesen. [4]

Die ursprüngliche Implementierung des Uber-Taxidienstes bestand aus einem monolithischen PHP-Server, der eingehende Anfragen verarbeitete und entsprechende Änderungen in einer MySQL-Datenbank vornahm. Mit dem kontinuierlichen Wachstum der Nutzerbasis traten jedoch zunehmend Probleme im Zusammenhang mit konkurrierenden Zugriffen auf. Diese äußerten sich unter anderem darin, dass Nutzer unbeabsichtigt mehrere Fahrten gleichzeitig buchten oder einzelne Fahrer mehreren Kunden gleichzeitig zugewiesen wurden. [5]

Im Jahr 2011 erfolgte die erste Aufteilung des ursprünglichen Server-Monolithen in zwei separate Dienste: den Dispatcher und den API-Service. Der Dispatcher war für die Zuordnung von Fahrgästen zu verfügbaren Fahrern verantwortlich und wurde in Node.js implementiert. Der Einsatz einer ereignisbasierten, asynchronen Ausführungsumgebung stellte gegenüber dem vorherigen Ansatz einen wesentlichen Vorteil dar. Der Dispatcher kommunizierte mit einem API-Service, der weiterhin einen Großteil der Geschäftslogik der Anwendung enthielt, einschließlich der Buchungs- und Zahlungsprozesse. Durch die Aufteilung in zwei Backend-Dienste erfolgte damit eine erste Annäherung an eine serviceorientierte Architektur, wenngleich die Architektur noch deutlich von einer späteren Microservice-Landschaft entfernt war. [3]

Mit dem weiteren Wachstum der Anwendung nahm auch der Aufwand für Build- und Bereitstellungsprozesse kontinuierlich zu. Änderungen an einzelnen Komponenten erforderten weiterhin die Neukompilierung und Bereitstellung der gesamten Anwendung. Parallel zum Wachstum der Nutzerbasis stieg auch die Anzahl der beteiligten Entwicklerinnen und Entwickler, wodurch sowohl Entwicklungs- als auch CI/CD-Prozesse zunehmend komplexer, kostenintensiver und zeitaufwändiger wurden. Da monolithische Anwendungen grundsätzlich als geschlossene Einheit gebaut und ausgeführt werden, konnten diese Herausforderungen innerhalb der bestehenden Architektur nur begrenzt adressiert werden. Dies machte eine grundlegende Neugestaltung der Systemarchitektur erforderlich. [5]

Im Jahr 2014 begann Uber mit der Migration zu einer Microservice-Architektur. Das zentrale Ziel bestand darin, die starke Kopplung einzelner Komponenten zu reduzieren, indem diese in eigenständige Dienste ausgelagert wurden. [3] Die Einführung von Microservices führte zu grundlegenden Veränderungen sowohl in der Softwarearchitektur als auch im Entwicklungsprozess. Im Gegensatz zu einer rein logischen Modularisierung innerhalb eines Monolithen ermöglicht die Architektur eine physische Trennung einzelner Verantwortungsbereiche. Darüber hinaus können die verschiedenen Dienste unabhängig voneinander entwickelt und in unterschiedlichen Programmiersprachen implementiert werden. Dies eröffnet die Möglichkeit, technologiespezifische Optimierungen vorzunehmen und die verfügbaren Hardwareressourcen effizienter auszunutzen. [3]

Gleichzeitig bringt die Microservice-Architektur gegenüber einem monolithischen Ansatz zusätzliche Herausforderungen mit sich. Insbesondere steigt die Komplexität des Gesamtsystems erheblich an. Dies betrifft nicht nur die Entwicklung der einzelnen Dienste, sondern auch Aspekte wie Sicherheit, Dienstkommunikation, Service-Discovery, Schnittstellenmanagement und Systemüberwachung. Im Bereich der Service-Discovery muss sichergestellt werden, dass eingehende Anfragen an die jeweils zuständigen Dienste weitergeleitet werden. Hierfür sind zusätzliche Mechanismen erforderlich, die die Lokalisierung und Adressierung der verfügbaren Dienste ermöglichen. Bezüglich Security muss exakt bestimmt werden, welche Dienste von welchen Quellen per Schnittstelle kontaktiert werden können, um unbefugten und nicht vorgesehenen Datenzugriff zu vermeiden. Ein Beispiel eines solchen sicherheitskritischen Dienstes ist ein Payment-Service, der mit sensiblen Zahlungsinformationen arbeitet.

Die Migration von monolithischen Systemen hin zu Microservice-Architekturen stellt ein Entwicklungsmuster vieler Anbieter großer verteilter Anwendungen dar. Ein weiteres bekanntes Beispiel hierfür ist Netflix. [4] Bei einem neu gegründeten Anbieter eines Taxidienstes ist daher zu erwarten, dass die Anwendung eine ähnliche Entwicklung erleben wird. Dadurch, dass der Dienstanbieter betriebswirtschaftlich zunächst eine finanzielle Grundlage aufbauen und die ersten Gewinnmargen ausreizen muss, erweitern sich damit entsprechend die technischen Möglichkeiten. Es ist daher riskant, das heutige Resultat des Wandels von Uber als eine allgemeine Blaupause für ein neu gegründetes Startup-Unternehmen zu betrachten. Für ein Startup-Unternehmen ist einzuschätzen, dass eine Microservice-Architektur einen zu hohen technischen und finanziellen Wartungsaufwand mit sich bringt, weshalb eine monolithische Architektur als eine natürliche, kostensparsame Alternative für die initiale Phase der Entwicklung folgt.

V. Techniken zur Gewährleistung der Skalierbarkeit

In den folgenden Unterkapiteln werden verschiedene Techniken vorgestellt, die zur Skalierbarkeit verteilter Anwendungen beitragen. Die meisten dieser Techniken sind jedoch nicht ausschließlich dem Fachgebiet der verteilten Systeme zuzuordnen, sondern stellen allgemeine Konzepte der Informatik dar, die in unterschiedlichen Arten von Softwareanwendungen Anwendung finden.

A. Zustandslosigkeit

Zustandslose Server zeichnen sich dadurch aus, dass die Verarbeitung einer Anfrage nicht vom internen Zustand der Anwendung abhängt. Vorangegangene Interaktionen beeinflussen das Ergebnis zukünftiger Interaktionen, beispielsweise in Form von Transaktionen oder API-Aufrufen, nicht unmittelbar. Jede Anfrage kann daher als isolierte Einheit betrachtet werden. Zustandslosigkeit führt zwar nicht zwangsläufig zu deterministischen Ergebnissen, da externe Zustandsquellen wie Datenbanken weiterhin Einfluss auf die Verarbeitung nehmen können, sie erhöht jedoch die Vorhersagbarkeit des Systemverhaltens. Der Grund hierfür liegt darin, dass kein innerhalb der Anwendung gespeicherter Zustand zur Beeinflussung der Anfrageverarbeitung herangezogen wird.

Da der betrachtete Taxidienst die Verarbeitung von Echtzeitdaten erfordert, kann nicht jede Komponente der Anwendung zustandslos realisiert werden. Ein naheliegendes Beispiel stellt die Echtzeitverfolgung eines bestellten Taxis dar. Um Positionsdaten in kurzen zeitlichen Intervallen an Nutzer zu übermitteln, ist eine Architektur erforderlich, die über klassische zustandslose Kommunikation per Anfrage und Antwort hinausgeht. Hierfür eignen sich insbesondere streamingbasierte Kommunikationsansätze, bei denen Daten kontinuierlich übertragen werden, anstatt ausschließlich über diskrete API-Aufrufe zu festgelegten Zeitpunkten ausgetauscht zu werden.

B. Geographische Skalierung durch Sharding

Uber bietet seine Dienste in zahlreichen Staaten und über mehrere Kontinente hinweg an. Daraus ergeben sich  große regionale Schwankungen hinsichtlich Bevölkerungszahl, Bevölkerungsdichte und Verkehrsaufkommen der bedienten Regionen. Unter diesen Voraussetzungen ist es nicht mehr ausreichend, sämtliche Nutzeranfragen unabhängig von ihrer geographischen Herkunft über dasselbe Rechenzentrum zu verarbeiten.

Sharding bezeichnet eine Skalierungstechnik, bei der die Gesamtheit der Daten und/oder Aufgaben einer Anwendung auf mehrere Server verteilt wird. [6] Die Architektur von Uber basiert stark auf dem Prinzip des Shardings, indem zahlreiche Prozesse, wie unter anderem Preisanpassung über Surge Pricing, auf Sharding-Strategien beruhen.

Das Anfrageaufkommen der Uber-Dienste unterscheidet sich erheblich zwischen einzelnen Regionen. In Großstädten ist mit einem deutlich höheren Verkehrsaufkommen zu rechnen als in ländlichen Gebieten. Darüber hinaus bestehen auch zwischen Staaten, Provinzen oder anderen Verwaltungseinheiten erhebliche Unterschiede hinsichtlich der durchschnittlichen Auslastung. Unter diesen Voraussetzungen ist es wenig zweckmäßig, sämtliche Anfragen über ein gemeinsames Rechenzentrum zu verarbeiten, da Regionen mit hohem Anfrageaufkommen die verfügbaren Ressourcen überlasten und dadurch auch Regionen mit geringerem Verkehrsaufkommen beeinträchtigen können.

In diesem Fall bietet es sich an, die verfügbaren Host-Systeme gezielt einzelnen geographischen Regionen zuzuordnen.

Da für die Nutzer eines Taxidienstes in erster Linie das unmittelbare geographische Umfeld relevant ist und nicht das weltweite Gesamtsystem, ist eine globale Verarbeitung sämtlicher Anfragen nicht erforderlich. Daraus ergibt sich die Möglichkeit, den geographischen Zuständigkeitsbereich einzelner Systemkomponenten gezielt zu begrenzen und die vorhandenen Hardwareressourcen auf bestimmte Regionen zu spezialisieren.

Für die Umsetzung von geographischem Sharding werden geographische Gebiete, beispielsweise Städte oder Metropolregionen, in einzelne Regionen beziehungsweise Zonen unterteilt, denen jeweils dedizierte Server zugeordnet sind. Jeder Server beziehungsweise jedes Servercluster übernimmt dabei die Verarbeitung der Anfragen einer bestimmten geographischen Region. Nutzer werden anhand ihres aktuellen geographischen Standorts einem entsprechenden Shard zugeordnet, wodurch eindeutig bestimmt werden kann, welche Server für die Verarbeitung ihrer Anfragen zuständig sind. Durch diese regionale Zuordnung können Hardwareressourcen bedarfsgerecht bereitgestellt werden, da die zuständigen Systemadministratoren den Ressourcenbedarf einzelner Regionen abschätzen und bei der Infrastrukturplanung berücksichtigen können.

Darüber hinaus trägt Sharding zur Ausfallsicherheit bei, da Sharding die Infrastruktur von Uber stark dezentralisiert. Ein Ausfall einer Stadt beziehungsweise einer geographischen Region soll keine anderen unverwandten Regionen auf der Welt beeinflussen.

Das Prinzip des geographischen Shardings findet sich auch außerhalb der Informatik. Ein anschauliches Beispiel stellt die Einteilung eines Staatsgebietes in Wahlbezirke dar. In westlichen Demokratien ist es üblich, dass aus jedem Wahlbezirk ein Abgeordneter in das nationale Parlament entsandt wird. Damit Wahlen möglichst gleichberechtigte Voraussetzungen für alle Wahlberechtigten und Parteien schaffen, wird angestrebt, dass die Wahlbezirke jeweils eine vergleichbare Anzahl an wahlberechtigten Einwohnern umfassen. Dieses Prinzip kann als politische Analogie zur gleichmäßigen Auslastung einzelner Regionen verstanden werden. Wahlbezirke gewährleisten darüber hinaus, dass regionale Interessen im Parlament repräsentiert werden, indem jede Region einen eigenen Abgeordneten wählt. Würde ein Staat hypothetisch nur aus einem einzigen Wahlbezirk bestehen, würde lediglich eine Partei den Wahlkreis gewinnen und dadurch mit einem einzigen Sitz zu 100% im Parlament vertreten sein, obwohl ihr tatsächlicher Stimmenanteil mit hoher Wahrscheinlichkeit deutlich unter 100% liegt. Die Unterteilung in mehrere Wahlbezirke ermöglicht somit eine differenzierte regionale Repräsentation. In Abbildung 1 ist eine Wahlbezirkskarte dargestellt, anhand der sich das Prinzip der regionalen Unterteilung eines Gesamtsystems veranschaulichen lässt. Dieses Grundprinzip lässt sich in weiten Teilen auf die Infrastruktur von Uber übertragen. Anstatt sämtliche Anfragen unabhängig von ihrer geographischen Herkunft durch dieselbe Infrastruktur zu verarbeiten und dadurch eine Überlastung einzelner Komponenten zu riskieren, werden Server beziehungsweise Servercluster gezielt bestimmten geographischen Regionen zugeordnet. Die Bereitstellung der Hardwareressourcen kann dadurch an das jeweilige regionale Anfrageaufkommen angepasst werden. Grundsätzlich bewirkt das Prinzip des Shardings daher, dass regionale Probleme mit regionalen, auf den konkreten Ort angepassten Mitteln gelöst werden.

Es muss beachtet werden, dass für diese Sharding-Strategie kein einheitlicher Konsens über den Namen herrscht. Sharding ist nur eine von mehreren Möglichkeiten, die Technik zu nennen, und bezeichnet das Einteilen von geographischen Regionen in Teilregionen zur effizienteren Zuweisung und Auslastung von Hardware. Sharding beantwortet lediglich die Frage, welcher Region welche technischen Ressourcen zugewiesen werden und welche anwendungsbezogenen Entitäten, wie beispielsweise Taxifahrer, für welchen Nutzer relevant sind. Andere Quellen können stattdessen unter anderem Begriffe wie Partitionierung oder Aufteilung verwenden.

Mit der Implementierung von geographischem Sharding ergeben sich mehrere weitere Fragestellungen. Hierzu gehören beispielsweise die Kriterien für die Bestimmung der räumlichen Größe und Anzahl der Regionen. Dabei soll insbesondere die Bevölkerungszahl und die regionale Nachfrage der Taxidienste berücksichtigt werden. Ebenfalls muss festgelegt werden, welche Bedingungen eine Neuzuweisung der regionalen Ausbreitung eines Shards voraussetzen. Darüber hinaus ist zu bestimmen, welche Daten ausschließlich innerhalb eines bestimmten Shards relevant sind, beispielsweise Echtzeitdaten laufender Taxifahrten, und welche Daten unabhängig von einer geographischen Region gespeichert werden können, wie etwa Reisehistorien oder Zahlungsinformationen von Nutzern. Schließlich stellt sich die Frage nach der Platzierung der Host-Server, die einem bestimmten Shard zugeordnet sind. Dabei muss berücksichtigt werden, dass die physische Infrastruktur womöglich nicht jederzeit am geographischen Standort des jeweiligen Shards zur Verfügung stehen wird.

Die Festlegung geeigneter geographischer Shards stellt insbesondere deshalb eine Herausforderung dar, weil zwischen städtischen und ländlichen Regionen unterschieden werden muss. Aufgrund des deutlich höheren Verkehrs- und Anfrageaufkommens in Großstädten bietet es sich an, diese als eigenständige Regionen zu behandeln, während das umliegende ländliche Gebiet zu einer separaten Region zusammengefasst wird. Da die Bevölkerungsdichte je nach Region erheblich variiert, ist es zweckmäßig, ländliche Gebiete in flächenmäßig größere Regionen einzuteilen als dicht besiedelte Ballungsräume. Besteht das Ziel darin, die Regionen so zu dimensionieren, dass sie jeweils eine vergleichbare Bevölkerungszahl umfassen, ergibt sich daraus zwangsläufig, dass ländliche Regionen eine deutlich größere geographische Ausdehnung besitzen als städtische Regionen mit hoher Bevölkerungsdichte. In Abbildung 2 wird dieses Prinzip anhand der administrativen Provinzeinteilung Algeriens veranschaulicht. Aufgrund der klimatischen Gegebenheiten weist Algerien eine stark ungleichmäßige Bevölkerungsverteilung auf. Während die nördliche Mittelmeerküste dicht besiedelt ist und entsprechend in zahlreiche flächenmäßig kleine Provinzen unterteilt wird, umfassen die zentralen und südlichen Landesteile überwiegend Wüstengebiete mit geringer Bevölkerungsdichte. Dementsprechend besitzen die südlichen Provinzen eine erheblich größere territoriale Ausdehnung als die Provinzen entlang der Mittelmeerküste. Dieses Muster entspricht dem bereits anhand der Wahlbezirke erläuterten Prinzip einer regionalen Unterteilung entsprechend der jeweiligen Bevölkerungsdichte.

Nicht alle Daten müssen einer konkreten geographischen Region zugeordnet werden. Je nach Art der gespeicherten Informationen eignen sich unterschiedliche Sharding-Strategien. Echtzeitdaten sollten grundsätzlich innerhalb des jeweiligen Shards gespeichert werden, da sie sich auf aktuelle Zustandsänderungen laufender Taxifahrten beziehen und überwiegend für die jeweilige Region relevant sind. Allgemeine Nutzerdaten, wie beispielsweise Namen oder Zahlungsinformationen, können hingegen unabhängig von einer geographischen Region gespeichert werden. Alternativ können sie aus Optimierungsgründen dem jeweiligen Shard eines Nutzers zugeordnet werden.

Darüber hinaus muss berücksichtigt werden, dass physische Server nicht zwangsläufig innerhalb der geographischen Region betrieben werden können, der sie logisch zugeordnet sind. Insbesondere kurzfristige Nachfragespitzen, beispielsweise infolge größerer Veranstaltungen, können dazu führen, dass die in einer Region verfügbaren Ressourcen nicht mehr ausreichen. Um in solchen Situationen nicht auf weit entfernte Host-Systeme zurückgreifen zu müssen, sollte die Infrastruktur in der Lage sein, innerhalb des weltweiten Serverbestands zunächst nach geographisch möglichst nahegelegenen verfügbaren Systemen zu suchen und diese gegebenenfalls einem neuen oder erweiterten Shard zuzuordnen.

C. Horizontale Skalierung

Neben der Umsetzung allgemeiner Optimierungsstrategien wie geographischer Zoneneinteilung, Load Balancing und der Optimierung von CPU-Laufzeiten ist zusätzlich ein Mechanismus erforderlich, der eine effektive horizontale Skalierung der zugrunde liegenden Hardware ermöglicht. Geographisches Sharding trägt zwar zur Entlastung der Gesamtinfrastruktur bei, indem Hardware-Ressourcen bestimmten geographischen Regionen zugeordnet werden und dadurch die Last entsprechend verteilt wird, jedoch reicht die reine Zoneneinteilung nicht aus, um Leistungseinbußen während Spitzenlastzeiten angemessen zu minimieren. Auch innerhalb eines einzelnen Shards können erhebliche Lastschwankungen auftreten. Daher wird ein Verfahren benötigt, das es ermöglicht, bei steigendem Ressourcenbedarf innerhalb kurzer Zeit zusätzliche Infrastruktur bereitzustellen und dem betroffenen Shard zuzuweisen.

Die Zustandslosigkeit der Server vereinfacht den Skalierungsprozess erheblich, da das Verhalten eines Dienstes nicht vom internen Zustand des jeweiligen Dienstes abhängt, sondern ausschließlich von externen Datenquellen, insbesondere relationalen Datenbanken. Dadurch kann leichter sichergestellt werden, dass mehrfach instanziierte Dienste ein konsistentes und vorhersagbares Verhalten aufweisen.

Uber verwendet hierzu einen Machine-Learning-basierten Algorithmus, die Capacity Recommendation Engine, der auf Grundlage historischer Daten Vorhersagen über den zukünftigen Kapazitätsbedarf trifft (Peak Throughput Estimation). Der Algorithmus fungiert somit als Mechanismus zur automatischen Skalierung und unterstützt sowohl die dynamische Erweiterung als auch die Reduzierung der verfügbaren Infrastrukturressourcen. [7]

Eine zentrale Herausforderung besteht darin, den erforderlichen Gesamtbestand an Hardware-Ressourcen so zu bestimmen, dass in allen Shards mit erhöhtem Ressourcenbedarf ausreichend Kapazitäten für Skalierungsmaßnahmen zur Verfügung stehen, ohne dass hierfür Ressourcen aus anderen Shards abgezogen werden müssen.

D. Vertikale Skalierung der CPU-Rechenzeit

Physische Server, auf denen Uber-Dienste ausgeführt werden, verfügen über Prozessoren mit einer festen Anzahl von CPU-Kernen. Da sowohl die Anzahl der gleichzeitig ausgeführten Dienste als auch deren Ressourcenbedarf erheblichen Schwankungen unterliegen, ist es aus Sicht der Ressourceneffizienz nicht sinnvoll, sämtlichen Diensten dauerhaft den Zugriff auf alle verfügbaren CPU-Ressourcen zu gewähren. Eine solche Überprovisionierung führt zu unnötigem Energieverbrauch und erhöhten Betriebskosten, obwohl ein erheblicher Teil der bereitgestellten Ressourcen zeitweise ungenutzt bleibt. [8]

Über einen längeren Zeitraum erfolgte die CPU-Provisionierung bei Uber manuell durch die zuständigen Systemadministratoren. Die für einzelne Dienste und deren Infrastruktur verantwortlichen Teams mussten die Ressourcenzuweisung regelmäßig anpassen, um auf Veränderungen der Systemauslastung reagieren zu können. [8]

Ziel der CPU-Provisionierung ist die bedarfsgerechte Zuweisung von CPU-Ressourcen an einzelne Container. Dabei sollen möglichst wenige Ressourcen reserviert werden, ohne die Leistungsfähigkeit oder Reaktionszeiten der Anwendung merklich zu beeinträchtigen. Dies setzt ein Verfahren voraus, das den tatsächlichen Ressourcenbedarf möglichst präzise bestimmen kann. [8]

Eine zentrale Herausforderung besteht darin, zunächst zu ermitteln, ob die verfügbare Hardware ausreichende CPU-Kapazitäten für einen bestimmten Workload bereitstellen kann und in welchem Umfang CPU-Ressourcen einem Container zugewiesen werden können. Die Ressourcenzuweisung erfordert daher eine vorgelagerte Analyse der verfügbaren Hardwarekapazitäten. Ohne eine entsprechende Planung besteht die Gefahr eines schwer vorhersehbaren Laufzeitverhaltens, da unterschiedliche Hardwarekonfigurationen bei einer Neuzuweisung zu abweichenden Leistungsmerkmalen führen können. [8]

Eine weitere Herausforderung ergibt sich aus den erheblichen Schwankungen der Systemauslastung im Verlauf einer Woche. Werden die zugewiesenen CPU-Ressourcen nicht regelmäßig und zeitnah an den tatsächlichen Bedarf angepasst, verbleiben Workloads über längere Zeiträume in einem überprovisionierten Zustand. Dies führt zu einer ineffizienten Nutzung der verfügbaren Infrastrukturressourcen und zu erhöhten Energiekosten beim Infrastrukturbetrieb. [8]

Darüber hinaus kann die Verlagerung von Workloads zwischen unterschiedlichen Rechnern einen erheblichen Zeitaufwand verursachen und unter Umständen mehrere Stunden in Anspruch nehmen. Aus diesem Grund sollten Strategien vermieden werden, die eine häufige Umverteilung von Ressourcen zwischen verschiedenen Systemen erfordern. [8]

E. Event-basierte Architektur

In einer ereignisgetriebenen Architektur (Event-driven Architecture) werden Funktionen nicht durch direkte Funktionsaufrufe innerhalb des Quellcodes ausgelöst. Stattdessen werden sie an bestimmte Ereignisse (Events) gekoppelt und durch den Empfang entsprechender Benachrichtigungen aktiviert. Die Ausführung einer Funktion erfolgt somit, sobald ein zugehöriges Event signalisiert, dass ein bestimmtes Ereignis eingetreten ist.

Ereignisgetriebene Architekturen dienen dazu, lose gekoppelte Dienste miteinander zu verknüpfen. Da die Verarbeitung von Ereignissen über definierte Event Listener erfolgt, können neue Verarbeitungsschritte hinzugefügt oder bestehende entfernt werden, ohne dass umfangreiche Änderungen am Quellcode eines Dienstes erforderlich sind. Dies trägt zur Entkopplung einzelner Systemkomponenten bei und erleichtert die Erweiterbarkeit der Gesamtarchitektur. [9]

Ein Event repräsentiert ein beliebiges Ereignis innerhalb der Anwendung. Im Kontext von Uber können dies beispielsweise die Erstellung eines Nutzerkontos oder die Buchung einer Taxifahrt sein.

Eine häufig eingesetzte Form der Event-Architektur ist die Publisher-Subscriber-Architektur. Ein Publisher erzeugt und veröffentlicht Ereignisse, während Subscriber diese Ereignisse empfangen und darauf reagieren. Die Kommunikation erfolgt dabei indirekt über eine vermittelnde Infrastruktur, sodass Publisher und Subscriber nicht unmittelbar voneinander abhängig sind. Dadurch können unterschiedliche Dienste dieselben Ereignisse verarbeiten und jeweils eigene Anwendungslogik ausführen. Ein Dienst kann dabei sowohl Ereignisse veröffentlichen als auch Ereignisse anderer Dienste verfolgen.

Für eine Anwendung wie Uber eignen sich ereignisgetriebene Architekturen insbesondere aufgrund der Anforderungen an die zeitnahe Verarbeitung von Zustandsänderungen. Beispiele hierfür sind die Übermittlung von Standortinformationen eines Fahrzeugs oder Statusaktualisierungen einer laufenden Fahrt. Obwohl es sich dabei in der Regel nicht um ein tatsächliches Echtzeitsystem handelt, sondern lediglich um Aktualisierungen in regelmäßigen Zeitintervallen, ermöglicht die Event-Architektur eine kontinuierliche Verteilung aktueller Informationen an die beteiligten Komponenten. Auf diese Weise kann die mobile Anwendung fortlaufend den aktuellen Zustand einer Fahrt oder die Position eines Fahrzeugs abrufen und darstellen.

Neben den Komponenten zur Erzeugung und Verarbeitung von Ereignissen ist zusätzlich ein Mechanismus zur Übermittlung der Events erforderlich. Die Einführung einer ereignisgetriebenen Architektur macht daher die Auswahl eines geeigneten Kommunikationsmodells notwendig. Dabei stellt sich insbesondere die Frage, ob Ereignisse als einzelne Nachrichten an die jeweiligen Empfänger übertragen werden sollen oder ob sie als kontinuierlicher Datenstrom bereitgestellt und dauerhaft gespeichert werden. Die Wahl des Kommunikationsmodells hat wesentlichen Einfluss auf Skalierbarkeit, Fehlertoleranz, Nachvollziehbarkeit von Ereignissen sowie die Möglichkeiten der späteren Datenanalyse.

Aufgrund der hohen Anforderungen bezüglich Echtzeit sowie der komplexen Prozesse bei einem Taxidienst eignet sich im gegebenen Anwendungsfall insbesondere das Streaming-Modell. Durch das kontinuierliche Übertragen von Events werden einzelne Ereignisse nach der Verarbeitung nicht verworfen, sondern es kann ein Protokoll eingegangener Ereignisse archiviert werden, wodurch bereits verarbeitete Events jederzeit abrufbar bleiben. Die Speicherdauer von Eventprotokollen kann zudem durch den Systemadministrator konfiguriert werden, wodurch der Speicheraufwand des Event-Stream-Ansatzes flexibel bleibt. Durch die kleinere Größe des Payloads bei Event Streams ist zusätzlich eine Verarbeitung viel größerer Mengen an Events pro Zeiteinheit möglich [10], was für einen verteilten Taxidienst, bei dem Taxibestellungen bereits pro Stunde häufig auftreten, von Vorteil ist. Apache Kafka ist ein Beispiel einer Technologie, die den Zweck erfüllt, Event-Streaming in einer Anwendung umzusetzen.

Die ereignisgetriebene Architektur ist im Kontext der Skalierbarkeit insofern relevant, dass durch diese Architektur der Entwicklungsaufwand auf der Seite des Unternehmens bedeutend sinkt, wodurch Dienste effektiver entwickelt, modifiziert und entfernt werden können. Das Gesamtnetzwerk aller betriebenen Dienste wird damit flexibler und leichter veränderbar, wodurch ein laufender Betrieb des Taxidienstes für die Nutzerseite leichter garantiert werden kann.

F. Load Balancing

Die Implementierung von geographischem Sharding und horizontaler Skalierung allein genügt nicht, um geringe Latenzen bei Nutzeranfragen sicherzustellen. Auch innerhalb eines einzelnen Shards kann es zu einer Überlastung der Backend-Dienste kommen. Dies kann beispielsweise auftreten, wenn ein Shard eine Großstadt repräsentiert und aufgrund einer größeren Veranstaltung eine erhöhte Nachfrage nach Taxifahrten entsteht. In solchen Situationen muss die Infrastruktur auch innerhalb des betroffenen Shards horizontal skaliert und die Netzwerklast gleichmäßig auf die verfügbaren Host-Systeme verteilt werden.

Die Verteilung einer großen Anzahl eingehender Anfragen auf mehrere Host-Systeme wird als Load Balancing bezeichnet. Load Balancing wird üblicherweise durch einen dedizierten Dienst oder eine spezialisierte Anwendung realisiert. Ziel ist es, einzelne Server zu entlasten, indem die Gesamtheit der eingehenden Nutzeranfragen auf mehrere Rechner verteilt wird. Dadurch wird nicht nur die Netzwerkauslastung einzelner Systeme reduziert, sondern auch eine effizientere Nutzung der verfügbaren Rechenressourcen erreicht. [11]

Mit dem Einsatz von Load Balancing werden insbesondere zwei Ziele verfolgt: die Steigerung der Zuverlässigkeit und die Verbesserung der Ressourceneffizienz. Werden Anfragen nicht gleichmäßig auf die verfügbaren Systeme verteilt, können einzelne Rechner überlastet werden, was zu erhöhten Antwortzeiten und damit zu einer Beeinträchtigung der Anwendungsleistung führt. Gleichzeitig wird die vorhandene Infrastruktur ineffizient genutzt, da potentiell zusätzliche Hardware bereitgestellt wird, obwohl bereits laufende Systeme noch über ungenutzte Kapazitäten verfügen. Load Balancing trägt daher dazu bei, die vorhandenen Ressourcen möglichst vollständig auszuschöpfen, bevor eine Erweiterung der Infrastruktur erforderlich wird.

Darüber hinaus erhöht Load Balancing die Ausfallsicherheit eines Systems. Fällt ein einzelner Rechner aus, können eingehende Anfragen weiterhin auf die verbleibenden Systeme verteilt werden. Die Folgen eines Ausfalls beschränken sich dadurch in der Regel auf eine moderate Verringerung der Gesamtleistung, ohne dass die Verfügbarkeit des Dienstes vollständig beeinträchtigt wird.

Grundsätzlich lassen sich zwei Formen des Load Balancings unterscheiden: statisches und dynamisches Load Balancing. Beim statischen Load Balancing erfolgt die Weiterleitung von Anfragen ausschließlich auf Grundlage vordefinierter Regeln, beispielsweise anhand der Reihenfolge der verfügbaren Server. Dynamisches Load Balancing berücksichtigt hingegen den aktuellen Zustand des Systems und trifft Entscheidungen auf Basis gegenwärtiger Metriken, wie etwa der aktuellen Auslastung oder der Antwortzeiten einzelner Server. [11]

Bei beiden Varianten ist es üblich, einzelnen Host-Systemen Gewichtungen zuzuweisen. Diese Gewichtungen ermöglichen eine gezielt ungleichmäßige Verteilung der Anfragen, beispielsweise wenn die beteiligten Systeme über unterschiedliche Hardwarekapazitäten verfügen. [11]

Im Kontext eines Taxidienstes eignet sich insbesondere dynamisches Load Balancing. Dieser Ansatz ermöglicht eine flexible Verteilung von Anfragen auch dann, wenn die Server eines Shards geographisch verteilt betrieben werden. Der zusätzliche Rechenaufwand des komplexeren Verfahrens führt dabei in der Regel nicht zu signifikanten Leistungseinbußen. Allerdings muss sichergestellt werden, dass die verwendeten dynamischen Algorithmen tatsächlich auf aktuelle Systemzustände reagieren und nicht faktisch dasselbe Verhalten wie einfache statische Verfahren aufweisen. Andernfalls könnte beispielsweise ein Algorithmus auf Basis der geringsten Antwortzeit effektiv dieselben Ergebnisse liefern wie ein Round-Robin-Verfahren, wodurch eine ineffiziente Nutzung der Rechenzeit des Load Balancers entstehen würde. [12]

Uber erweitert das Konzept des Load Balancings zusätzlich durch die Technik des Subsetting. Hierbei wird festgelegt,  welchen Teilmengen verfügbarer Dienste eingehende Anfragen verteilt werden. Ziel dieses Verfahrens ist die Reduzierung der Anzahl notwendiger Netzwerkverbindungen innerhalb des Dienstnetzes. Bei einer Verteilung sämtlicher Anfragen über alle verfügbaren Instanzen müssten zwischen den beteiligten Diensten und Proxys eine sehr große Anzahl von TCP-Verbindungen aufgebaut und verwaltet werden. Diese Zahl kommt durch die potentiell hohe Anzahl (häufig bis zu mehreren zehntausend) an Microservice-Containern, die auf einem gegebenen Host-Rechner zu einem Zeitpunkt ausgeführt wird. Mit einer zunehmenden Anzahl von Services, die entweder durch die Einrichtung neuer Dienste oder durch die horizontale Skalierung bestehender Services entsteht, wird das Problem der Anzahl von TCP-Verbindungen zunehmend signifikant. Ein Subset bezeichnet dabei lediglich eine feste Zahl von Services, die in einer logischen Gruppe zusammengefasst sind und einem Load Balancer zugewiesen werden. [12]

Netzwerkverbindungen zwischen Diensten werden in der Regel nicht für jede einzelne Anfrage neu aufgebaut und anschließend wieder geschlossen. Stattdessen werden bestehende Verbindungen über längere Zeiträume aufrechterhalten und für mehrere Anfragen wiederverwendet. Dieses Vorgehen reduziert den Kommunikationsaufwand erheblich, da insbesondere bei verbindungsorientierten Protokollen wie TCP der wiederholte Verbindungsaufbau zusätzlichen Overhead verursacht. Durch Subsetting kann die Anzahl der dauerhaft aufrechtzuerhaltenden Verbindungen begrenzt werden, wodurch das interne Service-Netzwerk entlastet wird. [12]

G. Skalierung von Datenbanken

Die bisherigen Abschnitte befassten sich ausschließlich mit der übergeordneten architektonischen Gestaltung eines verteilten Taxidienstes bezüglich der Anwendung als Gesamteinheit. Neben der Optimierung der Antwortzeit der Anwendung auf Nutzeranfragen stellt die Datenschicht jedoch eine weitere elementare Schicht eines verteilten Dienstes dar, die separat betrachtet werden muss. Durch die Signifikanz und Komplexität der persistenten Datenspeicherung bei verteilten Anwendungen existieren besondere Skalierungstechnicken, die speziell auf Datenspeicherung zugeschnitten sind. Neben der Optimierung der Antwortzeit von Microservices, die vom Endnutzer wahrgenommen wird, müssen ebenfalls die Reaktionszeiten der Datenschicht innerhalb der Anwendung optimiert werden, da diese auf ähnliche Weise anwendungsintern zu Verzögerungen bei der Bearbeitung einer Nutzeranfrage führen können.

Uber hat eine eigenständige Lösung zur Datenpersistierung mit dem Namen Docstore entwickelt. Docstore dient als eine allgemeine Lösung für mehrere datenbezogene Probleme, darunter Replizierung, Partitionierung, und Concurrency beim verteilten Datenbanksystem von Uber. [13]

Bei Uber werden mehrere Datenbankinstanzen innerhalb von Partitionen zusammengefasst. Dieses Prinzip ähnelt dem Prinzip des Shardings, welches zuvor beschrieben wurde. Eine Möglichkeit besteht darin, jede Datenbankinsatnz mit allen Tabellen auszustatten, die in der Datenbankarchitektur der Anwendung existieren, während die einzelnen Datensätze auf mehrere Partitionen und Datenbankinstanzen verteilt werden.

Ein häufiges Problem bei verteilten Datenbanken ist die Sicherstellung der Konsistenz. Uber löst dieses Problem mit dem Konsensalgorithmus Raft. Innerhalb einer Partition wird eine konkrete Datenbankinstanz als die primäre Datenquelle designiert und übergibt Datenaktualisierungen an die übrigen Instanzen innerhalb derselben Partition. Die primäre Datenquelle wird im Kontext von Raft als bezeichnet, während die übrigen Instanzen entsprechend als bezeichnet werden. Dementsprechend finden alle Schreib-Queries zunächst auf der Primärdatenbank statt. Im Anschluss wird der neue Datenzustand an die Follower übertragen.

Ein weiteres Problem bei verteilten Datenbanken ist die Durchführung von Join-Operationen zwischen mehreren Tabellen. In der Abfragesprache SQL existiert eine integrierte Syntax, die es ermöglicht, eine neue Tabelle zu erzeugen, die auf der logischen Schnittmenge zwei verschiedener Tabellen basiert. Bei der räumlichen Verteilung logisch zusammenhängender Daten über mehrere Instanzen ist eine einfache, lokale SQL-Abfrage nicht mehr ausreichend, um die gewünschten Daten vollständig abzurufen. [14]

Ebenfalls zu berücksichtigen ist die Auffindbarkeit eines bestimmten Datensatzes. Bei der Verteilung von logisch zusammenhängenden Daten über mehrere Partitionen und Datenbankinstanzen muss dennoch jederzeit die Möglichkeit bestehen, alle Daten einer fachlichen Entität, wie beispielsweise eines Nutzers, einer Taxifahrt, oder einer Bestellung, mit möglichst wenigen Queries abzufragen. Ein Lösungsweg besteht darin, das Prinzip der Primary Keys auf Partitionen anzuwenden. In SQL ist es geläufig, einzelnen Datensätzen einen Primary Key zuzuordnen, der als ein numerischer Identifikator für einen Datensatz agiert. Dieser Primary Key kann darauffolgend in Datenbankabfragen zur eindeutigen Identifizierung eines Datensatzes verwendet werden. Auf diese Weise wird auf alle Daten zugegriffen, die mit einer bestimmten Entität zusammenhängen. Dieses Prinzip lässt sich ebenso auf Datenbankpartitionen übertragen, sodass in einer Abfrage nicht nur der Primary Key einer anwendungsbezogenen Entität enthalten ist, sondern auch der Identifikator der Partitionen, in der die gesuchten Daten aufzufinden sind. [14] In diesem Fall bietet es sich an, zusätzlich eine Konvention zu etablieren, anhand der die Primary Keys für Partitionen benannt werden, sodass aus jedem Primary key die eindeutige geographische und logische Zone eines Datensatzes auszulesen ist.

Für die richtige Weiterleitung einer Datenbankabfrage an die relevanten Partitionen existiert bei Docstore eine Trennung in eine Query-Schicht und eine Speicherschicht. Die Query-Schicht ist dafür zuständig, die Abfragen nach ihrer Plausibilität zu prüfen und die Abfrage an die zugehörigen Partitionen weiterzuleiten. Auf der Speicherschicht findet die Partitionierung der Instanzen statt. Ebenfalls ist die Speicherschicht für Konsens und Replizierung verantwortlich. [15] Abbildung 3 veranschaulicht die Aufteilung der Schichten in einer grundlegenden Form.

H. Adaptive Preisregulierung durch Surge Pricing

Ein weiterer Aspekt der Skalierung betrifft die Kostenstruktur eines Systems. Diese beziehen sich nicht ausschließlich auf die Infrastrukturkosten, die bereits im Zusammenhang mit Load Balancing und vertikaler CPU-Skalierung berücksichtigt wurden, sondern umfassen auch betriebswirtschaftliche Optimierungspotenziale. In diesem Kontext können dynamische Preisanpassungen ebenfalls als eine Form der Skalierungsmaßnahme betrachtet werden.

Im Fall von Uber wird mittels Surge Pricing eine dynamische Anpassung der Fahrpreise in Abhängigkeit von der aktuellen Nachfrage vorgenommen. Dieser Prozess erfolgt algorithmisch. Surge Pricing stellt im Wesentlichen eine Mechanik zur Herstellung eines Gleichgewichts zwischen Angebot und Nachfrage dar, ohne dass eine manuelle Anpassung der Preise in einem dynamischen Umfeld wie dem Taximarkt erforderlich wird. Durch die Automatisierung können Preise in kurzen zeitlichen Intervallen, beispielsweise im Minutentakt, an die aktuelle Nachfrage angepasst werden. [16]

Ohne die Implementierung von Surge Pricing würden dem Unternehmen potenzielle Einnahmen entgehen, die sich aus dem dynamischen Verhältnis von Angebot und Nachfrage auf dem Markt ergeben. Darüber hinaus schafft eine dynamische Preisgestaltung Anreize für Fahrdienstpartner, insbesondere in Phasen hoher Nachfrage (hoher Surge-Faktoren) Fahrten anzubieten. In diesen Zeiträumen ist das Einkommen der Fahrer erhöht, während die Nachfrage der Kunden weiterhin ausreichend hoch bleibt, sodass trotz steigender Preise weiterhin eine signifikante Anzahl an Fahrten zustande kommt. [16]

Die technische Umsetzung von Surge Pricing basiert auf Event-Streaming. Zur Berechnung eines erhöhten Surge-Faktors wird der vorliegende Event-Stream analysiert, wobei die Dichte der vergangenen Taxibestellungen bestimmt wird. Auf dieser Grundlage wird ein Faktor berechnet, der mit dem Standardpreis multipliziert wird. Die Preisberechnung ist somit direkt von der Historie der erfassten Bestell-Events abhängig. Der Surge-Faktor kann sich bei Uber innerhalb kurzer Zeiträume mehrfach ändern, typischerweise etwa alle fünf Minuten, wodurch eine robuste Event-Architektur erforderlich ist, die schnelle Reaktionszeiten ermöglicht.

Surge Pricing erweitert die Anforderungen an eine eventgetriebene Architektur dahingehend, dass Events nicht ausschließlich einmalig an andere Services weitergegeben werden, sondern auch nach ihrer Verarbeitung Auswirkungen auf nachgelagerte Prozesse haben können, da sämtliche relevanten Ereignisse im System potenziell Einfluss auf zukünftige Preisberechnungen nehmen. [16]

Darüber hinaus verfügt Surge Pricing über ein eigenes System zur Einteilung von Regionen in logische Einheiten. Während beim geographischen Sharding diese Einteilung primär der Optimierung der Infrastruktur dient, werden Regionen beim Surge Pricing zur differenzierten Preisgestaltung auf Basis der regionalen Nachfrage verwendet. Im Gegensatz zum geographischen Sharding ist dieser Prozess jedoch komplexer, da die Preiszonen dynamisch von der Nachfrage abhängen, welche innerhalb eines Tages erheblichen Schwankungen unterliegen kann. Das System des geographischen Shardings weist diese dynamische Abhängigkeit nicht auf. Folglich können sich die Regionsgrenzen im Surge-Pricing-Kontext mehrfach täglich verschieben, was in der Regel den Einsatz automatisierter Verfahren erforderlich macht. [16]

VI. Fazit

Mit den vorgestellten Skalierungstechniken werden verschiedene Ziele verteilter Systeme verfolgt, wobei zugleich unterschiedliche Zielkonflikte und Kompromisse berücksichtigt werden müssen. Geographisches Sharding gewährleistet einen effizienten Einsatz der verfügbaren Hardware, indem jeder Region entsprechend ihres Anfrageaufkommens ausreichende Ressourcen zugewiesen werden. In Kombination mit der dynamischen Anpassung der Hardwarekapazitäten pro Shard wird damit das Prinzip der horizontalen Skalierung umgesetzt.

Zur Gewährleistung einer losen Kopplung sowie einer hohen Austauschbarkeit einzelner Dienste wird eine ereignisgetriebene Architektur eingesetzt. Diese Architektur stellt eine Schnittstelle bereit, über die Funktionsaufrufe durch das Auslösen von Ereignissen initiiert werden, sodass Funktionen nicht unmittelbar im Quellcode anderer Dienste aufgerufen werden müssen. Dadurch kann das Verhalten einzelner Dienste in hohem Maße angepasst werden, ohne umfangreiche Änderungen am Quellcode vorzunehmen. Darüber hinaus erleichtert dieser Ansatz sowohl den Austausch bestehender Dienste als auch das Hinzufügen oder Entfernen weiterer Komponenten innerhalb der Gesamtarchitektur, da hierfür lediglich geringfügige Anpassungen an den verknüpften Event Listenern erforderlich sind. Mit dieser Technik wird eine Skalierbarkeit im Entwicklungsprozess sichergestellt.

Für die vertikale Skalierung der Hardware eignet sich die dynamische Zuweisung von CPU-Ressourcen. Analog zum geographischen Sharding wird dabei sichergestellt, dass die verfügbaren CPU-Kerne möglichst effizient genutzt werden, indem jedem Workload nur die tatsächlich benötigte Rechenkapazität zugewiesen wird. Eine effiziente Nutzung der CPU-Ressourcen trägt zugleich zur Senkung der Betriebskosten bei, da der Energieverbrauch der Infrastruktur reduziert werden kann. Gleichzeitig bleiben jedoch Herausforderungen bestehen, insbesondere die möglichst präzise Bestimmung des tatsächlichen Ressourcenbedarfs eines Workloads sowie die Auswahl geeigneter CPU-Ressourcen unter Berücksichtigung unterschiedlicher Hardwarekonfigurationen.

Neben technischen und infrastrukturellen Fragestellungen muss auch die betriebswirtschaftliche Perspektive eines Taxidienstes berücksichtigt werden. Während durch die Optimierung der Ressourcennutzung Betriebskosten reduziert werden können, ermöglicht eine dynamische Preisgestaltung zusätzlich eine Optimierung der Einnahmen. Mithilfe eines Surge-Pricing-Algorithmus können Taxipreise mehrmals innerhalb einer Stunde an das aktuelle Verhältnis von Angebot und Nachfrage angepasst werden. An diesem Beispiel wird das Zusammenspiel verschiedener Skalierungstechniken ersichtlich: Die ereignisgetriebene Architektur ermöglicht die kontinuierliche Protokollierung von Bestellungen und stellt die Grundlage für die Berücksichtigung vergangener Ereignisse bei der Neuberechnung der Fahrpreise dar. Gleichzeitig spielt auch die geographische Einteilung eine wesentliche Rolle, da Angebot und Nachfrage regional unterschiedlich ausfallen und sich folglich für jede Region ein eigenes Marktgleichgewicht mit entsprechend unterschiedlichen Fahrpreisen ergibt.

Die betrachteten Skalierungstechniken verfolgen überwiegend das Ziel, die Leistungsfähigkeit und Ausfallsicherheit verteilter Anwendungen zu erhöhen sowie Concurrency-Fehler zu vermeiden. Gleichzeitig führen sie jedoch zu einer höheren Komplexität bei der Entwicklung, dem Betrieb und der Wartung der Anwendung. Insgesamt zeigt sich, dass zahlreiche Skalierungsstrategien auf einer gezielten räumlichen oder logischen Aufteilung von Verantwortlichkeiten beruhen, sowohl hinsichtlich der Infrastruktur als auch der Softwarearchitektur. Daraus lässt sich schließen, dass die Skalierung verteilter Anwendungen maßgeblich auf einer klaren Zuordnung von Rollen, Verantwortlichkeiten und Ressourcen basiert, sodass der Betrieb eines Taxidienstes als verteilte Anwendung unabhängig von der regionalen Ausbreitung und Nachfrage der Angebote jederzeit stabil bleibt.

Referenzen

[1] M. van Steen and A. S. Tanenbaum, Distributed Systems, 4th ed., Jan.
2025.

[2] Y. Fu and C. Soman, “Real-time Data Infrastructure at Uber,” in
Proceedings of the 2021 International Conference on Management
of Data, Jun. 2021, pp. 2503–2516, arXiv:2104.00087 [cs]. [Online].
Available: http://arxiv.org/abs/2104.00087

[3] J. Clemm, “A Brief History of Scaling Uber,” Jan. 2024. [Online]. Available: https://www.joshclemm.com/writing/a-brief-
history-of-scaling-uber/

[4] C. Harris, “Microservices vs. monolithic architecture | Atlassian.” [On-
line]. Available: https://www.atlassian.com/microservices/microservices-
architecture/microservices-vs-monolith

[5] E. Haddad, “Service-Oriented Architecture: Scaling the Uber
Engineering Codebase As We Grow,” Sep. 2015. [Online]. Available:
https://www.uber.com/de/en/blog/service-oriented-architecture/

[6] S. Lee, Z. Guo, O. Sunercan, J. Ying, T. Kooburat, S. Biswal,
J. Chen, K. Huang, Y. Cheung, Y. Zhou, K. Veeraraghavan, B. Damani,
P. M. Ruiz, V. Mehta, and C. Tang, “Shard Manager: A Generic
Shard Management Framework for Geo-distributed Applications,” in
Proceedings of the ACM SIGOPS 28th Symposium on Operating
Systems Principles, ser. SOSP ’21. New York, NY, USA: Association
for Computing Machinery, 2021, pp. 553–569. [Online]. Available:
https://dl.acm.org/doi/10.1145/3477132.3483546

[7] S.-M. Peng, R. Dey, and J. He, “Capacity Recommendation Engine:
Throughput and Utilization Based Predictive Scaling.” [Online].
Available: https://www.uber.com/de/en/blog/capacity-recommendation-
engine/

[8] L. Vilhelmsen, “Vertical CPU Scaling: Reduce Cost of
Capacity and Increase Reliability,” Jul. 2022. [Online]. Available:
https://www.uber.com/us/en/blog/vertical-cpu-scaling/

[9] “What is event-driven architecture?” Sep. 2019. [Online].
Available: https://www.redhat.com/en/topics/integration/what-is-event-
driven-architecture

[10] J. Ferrer, “Kafka vs RabbitMQ: Key Differences & When to Use Each,”
Feb. 2025. [Online]. Available: https://www.datacamp.com/blog/kafka-
vs-rabbitmq

[11] “Types of load balancing algorithms.” [Online]. Avail-
able: https://www.cloudflare.com/learning/performance/types-of-load-
balancing-algorithms/

[12] C.-C. Liao, S. Kundu, and P. Krolikowski, “Better Load
Balancing: Real-Time Dynamic Subsetting.” [Online]. Avail-
able: https://www.uber.com/de/en/blog/better-load-balancing-real-time-
dynamic-subsetting/

[13] H. Chaudhary, O. Tariq, and D. Chatterjee, “Evolving Schemaless
into a Distributed SQL Database,” Feb. 2021. [Online]. Available:
https://www.uber.com/us/en/blog/schemaless-sql-database/

[14] A. Ahmad, “Scaling SQL Databases: 8 Challenges of
Horizontally Scaling SQL Databases.” [Online]. Available:
https://www.designgurus.io/blog/scaling-sql-databases

[15] D. Vaidya, P. Deshpande, and M. Ma, “How Uber
Conquered Database Overload: The Journey from Static Rate-
Limiting to Intelligent Load Management,” Apr. 2026. [Online].
Available: https://www.uber.com/us/en/blog/from-static-rate-limiting-to-
intelligent-load-management/

[16] L. Chen, A. Mislove, and C. Wilson, “Peeking Beneath the
Hood of Uber,” in Proceedings of the 2015 Internet Measurement
Conference, ser. IMC ’15. New York, NY, USA: Association
for Computing Machinery, 2015, pp. 495–508. [Online]. Available:
https://dl.acm.org/doi/10.1145/2815675.2815681