优步致力于在我们的全球市场上提供更安全和更可靠的运输。为了实现这一目标，优步依赖于在每个级别进行数据驱动的决策预测骑手需求在高流量事件期间识别和解决瓶颈在我们的司机合作伙伴注册过程中。随着时间的推移，对更多洞察力的需求导致了超过100个PB的分析数据，需要通过基于Hadoop的大数据平台进行最小延迟。自2014年以来，我们已致力于开发一个大数据解决方案，可确保数据可靠性，可扩展性和易用性，现在旨在提高我们平台的速度和效率。

在本文中，我们潜入了优步的Hadoop平台之旅，并讨论了我们在扩大这种丰富和复杂的生态系统方面建立的建设。

一代1：超级数据的开始

在2014年之前，我们有限的数据可以融入一些传统的在线事务处理（OLTP）数据库（在我们的案例中，MySQL和PostgreSQL）。要利用此数据，我们的工程师必须单独访问每个数据库或表，如果需要将数据组合使用来自不同数据库的数据，则留给用户编写自己的代码。此时，我们没有全局访问或全局访问我们所有存储的数据。实际上，我们的数据分散在不同的OLTP数据库中，总数据大小在几个Tberytes的顺序中，访问此数据的延迟非常快（通常，分钟）。图1如下，概述了2014年之前我们的数据架构：

随着优步的业务呈指数级增长(无论是从我们运营的城市/国家的数量，还是从每个城市使用该服务的乘客/司机数量来看)，传入数据的数量也增加了，需要在一个地方访问和分析所有数据，这要求我们构建第一代分析数据仓库。为了让优步尽可能受数据驱动，我们需要确保分析数据能够被分析师获取，而且都在一个地方。为了实现这一目标，我们首先将数据用户分为三个主要类别:

1. 城市运营团队（成千上万的用户）：这些工作人员负责管理和扩大优步在每个市场的交通网络。随着我们的业务扩展到新城市，有数千个城市运营团队定期访问这些数据，以应对司机和乘客的特定问题。
2. 数据科学家和分析师（数百名用户）：这些是分析师和科学家遍布不同的功能群体，需要数据，以帮助为用户提供最佳的运输和交付经验，例如预测骑士要求对未来的服务我们的服务。
3. 工程团队（数百名用户）：该公司的工程师专注于建立自动化数据应用，例如我们的欺诈检测和驾驶员船上平台。

第一代我们的分析数据仓库专注于将所有优步数据聚集在一个地方以及简化的数据访问中。对于前者，我们决定使用vertica作为我们的数据仓库软件，因为它的快速，可扩展和面向列的设计。我们还开发了多个Ad Hoc ETL（提取，转换和加载）作业，该作业将数据从不同的源（即AWS S3，OLTP数据库，服务日志等）复制到Vertica中。为了实现后者，我们将SQL标准化为我们的解决方案的接口，并建立了在线查询服务以接受用户查询并将其提交给底层查询引擎。图2，下面，描绘了此分析数据仓库：

发布我们的第一个数据仓库服务对于整个公司的工程师来说都是巨大的成功。首次，用户拥有全局视图，可以在一个地方访问所有数据。这导致了大量使用数据分析作为其技术和产品决策的基础。在几个月内，我们的分析数据的规模增长到数十岁，用户数量增加到数百个。

使用SQL作为一个简单的标准接口，使城市运营商能够轻松地与数据交互，而不知道底层技术。此外，不同的工程团队开始构建服务和产品的产品，这些数据由此数据通知（即Uberpool，Upfront定价等）和新团队形成为更好地使用并提供此数据（即，我们的机器学习和实验团队）。

限制

另一方面，我们的数据仓库和传入数据的广泛使用暴露了一些局限性。由于数据是通过临时的ETL作业获取的，而且我们缺乏正式的模式通信机制，因此数据可靠性成为一个问题。我们的大部分源数据都在json格式而摄入作业并不适合制作人代码的变化。

随着我们公司的增长，我们的数据仓库越来越昂贵。为了减少成本，我们开始删除旧的，过时的数据，以释放新数据的空间。在此之上，我们的大部分大数据平台都没有水平扩展，因为主要目标是取消阻止关键业务需求所需的集中数据访问或视图，并且根本没有足够的时间来确保所有部件都是水平可扩展的。我们的数据仓库有效地用作数据湖，堆积所有原始数据以及执行所有数据建模和服务。

此外，由于缺乏生产数据和下游数据消费者的服务之间缺乏正式合同（使用灵活的JSON格式导致缺乏架构实施）的服务之间缺乏正式合同，将数据摄入数据的ETL作业也非常脆弱。源数据）。如果不同的用户在摄入期间执行不同的变换，则可以多次摄取相同的数据。这导致我们上游数据源（即，在线数据存储）额外压力并影响了他们的服务质量。此外，这导致多个几乎相同数据存储在我们的仓库中，进一步提高了存储成本。在数据质量问题的情况下，回填是非常时间和劳动的，因为ETL作业是临时和源相关的，并且在摄入期间进行数据预测和转换。由于我们的摄取工作中缺乏标准化，因此难以摄取任何新数据集和类型。

二生2：Hadoop的到来

为解决这些限制，我们在Hadoop生态系统周围重新归建我们的大数据平台。更具体地说，我们介绍了一个Hadoop数据湖，其中所有原始数据都仅从不同的在线数据存储一次并且在摄入期间没有转换。这种设计转变显着降低了在线数据存储上的压力，并使我们从Ad Hoc Engestion Job转换到可扩展的摄取平台。为了让用户访问Hadoop中的数据，我们介绍了普拉斯托启用交互式Ad Hoc用户查询，阿帕奇火花促进编程访问原始数据（以SQL和非SQL格式），以及Apache Hive.作为非常大的查询的主力。这些不同的查询引擎允许用户使用最能解决其需求的工具，使我们的平台更灵活可访问。

为了保持平台可扩展，我们确保了所有在Hadoop中发生的数据建模和转换，在问题出现时启用快速回填和恢复。只有最关键的建模表（即，即将运行纯粹的快速SQL查询的城市运营商杠杆的最关键的模型表（即，那些杠杆刚刚达到的人）被转移到我们的数据仓库中。这显着降低了运行巨大数据仓库的运营成本，同时也将用户指导以与其特定需求为设计的基于Hadoop的查询引擎。

我们还利用了标准的柱状文件格式阿帕奇菌落，导致储存节省，给出了Ground压缩比和计算资源增益，给出了用于服务分析查询的柱状访问。此外，Parquet与Apache Spark的无缝集成使得该解决方案成为访问Hadoop数据的流行选择。图3，下面，总结了我们第二代大数据平台的架构：

除了合并Hadoop数据湖之外，我们还在该生态系统中进行了所有数据服务，水平可扩展，从而提高了我们大数据平台的效率和稳定性。特别是，拥有这种通用的水平可扩展性以解决即时业务需求，使我们能够将我们的能量集中在建立下一代数据平台上，而不是临时问题解决。

与我们的第一代数据流水线易受上游数据格式发生变化的平台不同，我们的第二次迭代使我们能够将所有数据的方式示思，从JSON转换为镶木地板以将架构和数据存储在一起。为实现此目的，我们构建了一个中央架构服务以收集，存储和服务模式以及不同的客户端库，以将不同的服务与此中央架构服务集成。脆弱，ad hoc数据摄入作业被替换为标准平台，以将其原始嵌套格式转移到Hadoop数据湖中的所有源数据。通过Hadoop中通过水平可扩展的批处理作业在摄入后发生的任何所需操作和转换。

随着优步的业务继续以光速扩展，我们很快就有了数十张PETABYTES的数据。我们的数据湖中增加了数十大的新数据，我们的大数据平台增长到超过10,000千年，在任何一天都有超过100,000个运行批量作业。这导致我们的Hadoop数据湖成为所有分析优步数据的集中源事。

限制

随着公司持续扩大缩放和储存在我们的生态系统中的数十张PETABYTES，我们面临着一系列新的挑战。

开始，存储在我们的HDF中的大量的小文件（由更多的数据被摄取，以及编写甚至更多输出数据的ad hoc批处理作业的更多用户）开始为HDFS Namenodes添加额外压力。最重要的是，数据延迟仍然远离我们所需的业务。每24小时只能向用户访问一次新数据，这太慢，无法进行实时决策。在将ETL和建模中移动到Hadoop的同时使此过程更加可扩展，因此这些步骤仍然是瓶颈，因为这些ETL作业必须在每次运行中重新创建整个建模表。添加到问题的内容，所有相关派生表的新数据和建模都是基于创建整个数据集的新快照，并交换旧表和新表，为用户提供对新数据的访问。摄入作业必须返回源数据存储，创建一个新快照，并在每次运行期间摄取或将整个数据集进行摄取或将整个数据集转换为COUNTABLE，CLANAR PALQUET文件。随着我们的数据商店的增长，这些工作可能需要20多小时，超过1,000多个火花执行者运行。

每个作业的大部分涉及从最新快照转换历史和新数据。虽然每个表中每天添加超过100千兆字节的新数据，但是每次运行都会为该特定表格转换整个，超过100 TB的数据集。对于ETL和建模作业，这也是如此，可以在每次运行中重新创建新的派生表。由于历史数据的更新比率高，这些作业必须依赖于基于快照的源数据的摄取。本质上，我们的数据包含大量的更新操作（即骑手和驾驶员评级或支持票价调整几个小时甚至几天）。由于HDFS和Parquet不支持数据更新，因此需要从更新的源数据创建新快照所需的所有摄取作业，将新快照摄取到Hadoop，将其转换为镶嵌格式，然后交换输出表以查看新数据。图4，下面，总结了这些基于快照的数据摄取如何通过我们的大数据平台移动：

第3代：长期重建我们的大数据平台

截至2017年初，我们的大数据平台被整个公司的工程和运营团队使用，使他们能够在一个地方访问新的和历史数据。用户可以轻松访问Hive，Presto，Spark，Vertica的数据，笔记本以及更多通过针对其需求量的单个UI门户网站提供更多仓库选项。在HDFS中拥有超过100个数据，我们的Compute集群中的100,000个vcores，每天100,000个Presto查询，每天10,000个火花作业，每天20,000个蜂巢查询，我们的Hadoop分析架构正在击打可扩展性限制，并且许多服务受到高位的影响数据延迟。

幸运的是，由于我们的底层基础设施水平可扩展以解决即时业务需求，我们有足够的时间来研究我们的数据内容，数据访问模式和用户特定要求，以确定建立下一代之前最紧迫的问题。我们的研雷竞技是骗人的究揭示了四个主要痛点：

1. HDFS可扩展性限制：这个问题面临着许多依赖HDFS扩展其大数据基础架构的公司。通过设计，HDFS是由其NameNode容量的瓶颈，因此存储大量小文件可以显着影响性能。当数据大小超过十个PETABYTES时，通常会发生这种限制，并且成为超过50-100 petabytes的真正问题。幸运的是，在几十多到几百个Petabytes中，将HDFS扩展了相对简单的解决方案，例如利用ViewFS和使用HDFS NameNode联合。通过控制小文件的数量并将数据的不同部分移动到单独的集群（例如，HBase和Yarn应用程序日志移动到单独的HDFS群集中），我们能够减轻这种HDFS限制。
2. Hadoop中的更快数据：优步的业务实时运行，因此，我们的服务需要访问尽可能新鲜的数据。因此，对于许多用例来说，24小时数据延迟太慢了，并且对更快的数据交付需求巨大。我们的第二代大数据平台的快照的摄入方法效率低下，防止了我们摄取了较低延迟的数据。为了加快数据传递，我们必须重新建立我们的管道，以增量摄入仅更新和新数据。
3. 支持Hadoop和Parquet中的更新和删除：Uber’s data contains a lot of updates, ranging in age from the past few days (e.g., a rider or driver-partner adjusting a recent trip fare) to a few weeks (e.g., a rider rating their last trip the next time they take a new trip) or even a few months (e.g., backfilling or adjusting past data due to a business need). With snapshot-based ingestion of data, we ingest a fresh copy of the source data every 24 hours. In other words, we ingest all updates at one time, once per day. However, with the need for fresher data and incremental ingestion, our solution must be able to support update and delete operations for existing data. However, since our Big Data is stored in HDFS and Parquet, it is not possible to directly support update operations on the existing data. On the other hand, our data contains extremely wide tables (around 1,000 columns per table) with five or more levels of nesting while user queries usually only touch a few of these columns, preventing us from using non-columnar formats in a cost-efficient way. To prepare our Big Data platform for long-term growth, we had to find a way to solve this limitation within our HDFS file system so that we can support update/delete operations too.
4. 更快的ETL和建模：类似于原始数据摄取，ETL和建模作业是基于快照的，需要我们的平台在每次运行中重建派生表。为了减少建模表的数据延迟，ETL作业也需要成为增量。这需要ETL作业仅逐步逐步释放从原始源表中的更改数据，并更新前一个派生的输出表，而不是每隔几个小时重建整个输出表。

介绍哈迪

考虑到以上要求，我们建成了Hadoop Upserts和增量(Hudi)，一个开源的Spark库，在HDFS和Parquet之上提供一个抽象层，以支持所需的更新和删除操作。Hudi可以在任何Spark作业中使用，是水平可扩展的，只依赖于HDFS来操作。因此，任何需要支持历史数据更新/删除操作的大数据平台都可以利用Hudi。

哈迪使我们能够在Hadoop中更新，插入和删除现有的镶嵌数据。此外，Hudi允许数据用户逐步逐步释放更改的数据，显着提高查询效率并允许派生建模表的增量更新。

我们的Hadoop生态系统中的原始数据基于时间分区，并且任何旧分区都可以在稍后的时间接收更新。因此，对于数据用户或依赖于这些原始源数据表的ETL作业，知道日期分区的唯一方式包含更新的数据是扫描整个源表并基于一些已知的时间概念过滤掉记录。这导致计算昂贵的查询需要完整的源表扫描，并防止常时运行ETL作业。

With Hudi, users can simply pass on their last checkpoint timestamp and retrieve all the records that have been updated since, regardless of whether these updates are new records added to recent date partitions or updates to older data (e.g., a new trip happening today versus an updated trip from 6 months ago), without running an expensive query that scans the entire source table.

使用哈迪库，我们能够远离基于快照的原始数据的摄取到增量摄取模型，使我们能够将数据延迟从24小时减少到不到一小时。图5是下面的，描绘了哈迪融入的大数据平台：

泛型数据摄取

哈迪不是第三代我们的大数据平台的补充。我们还通过Apache Kafka正式地形成了存储和大数据团队之间的上游数据存储区的变更。上游数据存储事件（以及来自不同应用程序和服务的经典记录消息）流入Kafka，统一的Avro编码包括附加标准全局元数据标题（即时间戳，行键，版本，数据中心信息和始发主机）。流媒体和大数据团队都使用这些存储变更活动作为其源输入数据以进行进一步处理。

我们的数据摄取平台，马尔马雷，以迷你批处理运行，从Kafka拾取上游存储变更日，将它们应用于Hadoop中的现有数据哈迪图书馆。如前所述，Hudi支持Upsert操作，允许用户添加新记录和更新或删除历史数据。摄入火花作业每10-15分钟运行每10-15分钟，在Hadoop提供30分钟的原始数据延迟（具有1-2个摄取工作故障或重试的余地。为了避免效率低下，由于超过一次将相同的源数据摄入到Hadoop中，我们的设置不允许在原始数据摄取期间的任何转换，从而导致我们决定使我们的原始数据摄取框架成为EL平台，而不是传统的ETL平台。在此模型下，鼓励用户在Hadoop内执行所需的转换操作，并在上游数据以其原始嵌套格式之后批量模式执行。

由于实现了对我们大数据平台的这些变化，因此通过避免不必要或低效的摄取操作，我们已经保存了大量的计算资源。我们的原始数据的可靠性也显着提高，因为我们现在可以避免摄入期间的错误变换。现在，用户可以使用任何大数据处理引擎在原始源数据顶部运行转换。此外，如果有任何问题，用户可以通过使用更多计算资源和更高程度的并行度来重新运行其转换并达到其SLA，以便更快地完成批量转换作业。

增量数据建模

考虑到需要摄取到Hadoop的大量上游数据存储（截至2017年超过3,000个Raw Hadoop表），我们还构建了一个通用的摄取平台，便于以统一和可配置的方式将原始数据摄入到Hadoop中。现在，我们的大数据平台逐步更新Raw Hadoop表，数据延迟为10-15分钟，允许快速访问源数据。但是，为了确保建模表也可用低延迟，我们必须在我们的建模ETL作业中避免效率低下（即全派生的表娱乐或完整源原始表扫描）。实际上，HUDI允许ETL作业仅从源表中获取更改的数据。建模作业只需要在每个迭代运行期间传递检查点时间戳到哈迪读者，以从原始源表接收新的或更新的记录（无论实际记录存储的日期分区）。

在ETL作业期间使用Hudi Writer使我们能够在派生建模表中更新旧分区，而不重新创建整个分区或表。因此，我们的建模ETL作业使用Hudi读者逐步获取来自源表的更改数据，并使用Hudi作家逐步更新导出的输出表。现在，ETL作业也在不到30分钟内完成，为Hadoop中的所有派生表提供不到1小时的端到端延迟。

为了提供具有不同选项的Hadoop表的数据用户访问所有数据或仅新的或更新的数据，使用Hudi存储格式的Hadoop Raw表提供了两种不同的读取模式：

1. 最新模式视图。在那个时间点，在整个Hadoop表上提供整体视图。此视图包括所有记录的最新合并值以及表中的所有现有记录。
2. 增量模式视图。仅根据给定的时间戳从特定Hadoop表中获取新的和更新的记录。此视图仅返回最近已插入或已更新的行以来最新检查点。此外，如果自上次检查点以来更新了一行以上的特定行，则此模式返回所有这些中间更改的值（而不是返回最新合并的值）

下面的图6描绘了所有存储在Hudi文件格式中的所有Hadoop表的两个读取视图：

用户通常根据自己的需要在这两个表视图之间切换。当它们运行一个临时查询以基于最新状态分析数据时，它们使用表的最新模式视图(例如，获取美国每个城市每周的总出行次数)。另一方面，当用户有一个迭代作业或查询，只需要获取最近一次执行以来的更改或新记录时，他们使用增量模式视图。这两个视图对所有Hadoop表在任何时候都可用，用户可以根据自己的需要在不同的模式之间切换。

标准化数据模型

除了提供相同表的不同视图之外，我们还标准化了我们的数据模型，为所有Raw Hadoop数据提供了两种类型的表：

1. changelog历史表。包含为特定上游表收到的所有变更的历史记录。此表使用户能够通过给定表格的更改历史扫描，并且可以为每个键合并以提供每行的最新值。
2. 合并快照表。存放上游表的最新合并视图。这个表包含每个键接收的所有历史变更日志的压缩合并视图。

下面的图7描绘了使用给定变更日的流为特定上游源数据存储生成不同的Hive Raw表：

然而，变更日志流可以或可能不包含给定密钥的整行（所有列）。虽然合并快照表始终为特定密钥提供所有列，但如果更改的上游流仅提供部分行更加常规，则更加亮起的变更历史表可能是稀疏的，这是通过避免仅在一个或几行重新发送整行时提高效率的功能限量列值已更改。如果用户想要从ChangElog历史表中获取更改的值并将其加入合并的快照表以创建完整行数据，我们还包括来自ChangElog历史表中的合并快照表中相同密钥的日期分区。这允许两个表通过避免在加入两者时避免合并的快照表的完整表扫描来更有效地加入特定分区。

图8，下面，总结了我们大数据平台的不同组件之间的关系：

第4一代：下一步是什么？

自2017年推出了第三代我们的大数据平台以来，本公司的用户可以快速可靠地访问Hadoop中的数据，但总有余地成长。下面我们总结了我们正在进行的努力，提升优步的大数据平台，以提高数据质量，数据延迟，效率，可扩展性和可靠性。

数据质量

为了提高数据质量，我们确定了两个改进的关键领域。首先，当一些上游数据存储不强制执行或检查存储之前，我们希望避免非模式符合数据的数据（例如，存储值是json blob的键值）。这导致输入我们Hadoop生态系统的坏数据，从而影响所有下游用户也依赖于此数据。为了防止涌入不良数据，我们正在转换到在数据内容上执行强制性模式检查的所有上游数据存储，并且如果有任何问题（例如，使用模式确认），则拒绝数据条目。

我们发现问题的第二个区域是实际数据内容的质量。在使用模式的同时，确保数据包含正确的数据类型，而且它们不检查实际数据值（例如，整数，而不是在[0,150]之间的正数）。为了提高数据质量，我们正在扩展我们的架构服务来支持语义检查。这些语义检查（换句话说，特定于优步数据类型）允许我们在基本结构类型检查之外的实际数据内容上添加额外的约束。

数据延迟

我们旨在将Hadoop的原始数据延迟减少到五分钟和建模表的数据延迟到十分钟。这将允许更多使用情况远离流处理，以更有效的迷你批处理处理，该处理使用Hudi的增量数据拉动。

我们还在扩展我们的Hudi项目来支持额外的视图模式，其中包括现有的读优化视图，以及新的实时视图，显示延迟仅几分钟的数据。此实时视图依赖于我们呼叫的开源解决方案（以及哈迪的一部分）合并读取或哈迪2.0。

数据效率

为了提高数据效率，我们正在远离依靠我们的服务和服务码头的专用硬件。此外，我们还统一我们的Hadoop生态系统中的所有资源调度程序，以弥合整个公司的Hadoop和非数据服务之间的差距。这允许以统一的方式计划的所有作业和服务如何，无论其介入是否可以执行。随着优步的增长，数据位置将是Hadoop应用程序的重要关注，并且一个成功的统一资源管理器可以将所有现有调度程序组合在一起(i.e., Yarn, Mesos, and Myriad).

可扩展性和可靠性

作为我们努力提高我们平台可扩展性和可靠性的一部分，我们确定了与可能的边缘案件相关的几个问题。虽然我们的摄入平台被开发为通用，可插拔模型，但上游数据的实际摄取仍然包括许多依赖性管道配置，使得摄取管道脆弱并增加经营数千个这些管道的维护成本。

为确保我们具有统一的数据摄取，无论数据源如何，我们都开始与优步数据存储团队合作的项目，以统一所有上游数据源的内容，格式和元数据，无论其技术化妆如何。该项目将确保有关这些特定上行技术的信息仅是添加到实际变更值的额外元数据（而不是为具有不同数据源的完全不同的变更内容和元数据），并且无论上游源如何发生数据摄取。

最后，我们的下一个版本的哈迪将允许我们在所有数据源的几分钟内默认地生成更大的镶木地板文件（与我们当前的128兆字节超过128兆字节）。它还将围绕更新与插入件的比率删除任何敏感性。Hudi 1.0依赖于一种呼叫的技术复制写作只要有更新的记录，它就会重写整个源片档案。这显着提高了写入放大，尤其是当inerted inerted的比率增加时，并阻止在HDFS中创建较大的镶木地板文件。胡迪的新版本is designed to overcome this limitation by storing the updated record in a separate delta file and asynchronously merging it with the base Parquet file based on a given policy (e.g., when there is enough amount of updated data to amortize the cost of rewriting a large base Parquet file). Having Hadoop data stored in larger Parquet files as well as a more reliable source-independent data ingestion platform will allow our analytical data platform to continue grow in the upcoming years as the business thrive.

向前进

优步数据组织是数据平台，数据基础，流和实时平台和大数据团队之间的跨功能合作，以构建支持优步的分析数据基础架构的所需库和分布式服务。如果在大量数据挑战中突破了突破了规模兴趣的限制，请考虑申请职务在我们的旧金山和帕洛阿尔托的团队。

请通过电子邮件将您的简历发送给hadoop-platform-jobs@uber.com.如果您有兴趣与我们合作！

Reza Shiftehfar目前领导优步拥有Hadoop平台团队。他的团队有助于构建和发展优化的可靠和可扩展的大数据平台，该平台提供利用Apache Hadoop，Apache Hive，Apache Kafka，Apache Spark和Presto等技术的Petabytes。Reza是优步数据团队的创始工程师之一，并帮助将Uber的数据平台从几个TB的数据平台到超过100个PEGABYTES，同时从24小时到几分钟的时间来降低数据延迟。