Uber实时数据处理架构

Kafka 2016 Summit上Uber工程师Danny Yuan分享了一个Streaming Processing PPT，如何解决Uber里Operation Team所需要的需求。看了整个视频觉得介绍很细致，这对于大部分LBS （Location Based Service）有很好的借鉴意义。

业务需求

Realtime OLAP

对于Operation部门而言，实时性很重要：

• 当前时间点，全球有多少量车在运行？有多少量车在空驶？
• 最近10分钟内，有多少UberX（类似于滴滴中的商务专车）在SF出现，热点地区在哪里？
• 每个区域的平均行驶时间、以及其他指标分别是多少？

作者给出了一个示意图，我们可以解读下：
1. 右侧是一个湾区的地图，通过蜂窝状六边形把坐标划分若干区域，红色就代表车的密集程度
2. 左侧是该区域在过去N分钟内各项指标的变化情况，例如平均的形式距离，接单率，平均客单价等
3. 通过筛选时间段、指标（Metric)等，可以全方面了解运营状况

这个图表让我相当了之前用TreeMap来监控集群利用率的场景，如出一辙。

• 左侧通过HeatMap显示各个机架上的不同时间段上Metric变化情况
• 右侧则是各指标在时间段上分布的场景

只不过在机器运维的Portal上显示的是，只不过我们面对的是集群，Uber面对的是车与地图：）

CEP（Complex Event Processing）复杂事件处理

例子：
1. 有多少个司机在最近10分钟内取消了3次接单以上？
2. 如果发现后，会通过聊天软件与司机对话

Supply Position 供求关系可视化

在什么位置供大于求，什么位置求大于供：

• 黄色的点代表需求
• 蓝色的店代表供应

处理的挑战

如何表示车辆的位置数据

地理位置函数，一般用得比较多的是GeoHash，通过切分空间的方法把二维坐标，转化成一维的数字。两个区间的比较查询，就演变成一个一维的比较函数。

Uber的做法正好相反，将坐标转化成一个特定的区域，通过六边形的办法来逼近真实的位置。使用六边形有这样几个好处：

1. 方便检索、查询、渲染
2. 容易找到周围相邻的邻居
3. 每个区域大小相同，形状相同

数据规模巨大

时间、空间、车辆状态、地理位置等组合会非常大

• 时间代表某一个时刻
• 空间在时间点上车的位置（例如LA，SF）
• 汽车的类型
• 状态（运行中，接单中，已接单出发地中等）

为了减少空间的规模：在地理位置、时间两个维度做了“取整”处理。通过六边形区域取整了地里位置，通过分钟级采样减少了其他状态，一天的数据量为

1dayofdata:300x10,000x7x1440x13=393billion

原始的数据为：

time, carID, locationX, locationY, status, .....

查询与计算的需求

1. 车的种类、状态非常多、因此查询场景是面向多维数据的。
2. 需要支持Heatmap，Top K， Histogram，count，avg，sum，percentage等计算函数
3. 巨大数据量：
   • 每秒百万级事件产生
   • 每个事件中有20+Field
4. 多种数据源
   • 司机端事件
   • 乘客端事件

Uber实时数据处理架构

分为5个部分：
1. 日志、事件数据来源框架 - Kafka
2. 数据清洗与处理，前置处理 - Samza
3. 存储系统 - Elastic Search
4. 数据读取，后置处理 - 自己开发的框架
5. 查询与构建与查询 - 自己构建
6. 应用层 - Web

数据采集与Kafka

这个Slides里面没有提到Uber架构，Google上找了一些相关的材料，整体架构如下：

数据来源有：

• Rider App
• Driver App
• API/Service（服务端）
• Dispatch （GPS 运行数据）
• Mapping & Logistic

日志、事件采集上在Kafka层包了Restful API，提供Java、Python、Go、NodeJS的SDK：

通过Samza进行清洗

主要有：
1. Transformation（坐标转化）：GPS坐标是二维的，为了能够根据城市和地域查询，转化成更离散化的数据：ZipCode、Hexagon（六边形坐标）、城市等。 (Lat, Long) -> (zipcode, hexagon, S2)
2. Pre-Aggregation：将一分钟数据归并成1分钟取整
3. Join Multiple Stream：例如Driver Status、Rider Status进行合并
4. Sessionization：将乘客的状态进行串联

From driver_canceled#window.time(10 min)
SELECT clientUUID, count(clientUUID) as cancelCount
GROUP BY clientUUID HAVING cancelCount > 3
INSERT INTO hipchat(room);

以上是一个ETL任务，每隔10分钟执行一次，既从Kafka中获得数据判断有问题的司机列表

通过这样的架构，支持运营人员能够在ES中清晰、索引的数据，获得实时分析能力：

同时由于在ES上层包装Query机制，也支持稍微复杂一些的离线查询。ES存储本身不是很好的离线方式，但对于离线查询频率不多的场景，也是够用的：

作者选型考虑

Lamdba vs Kappa

最终使用了Lamdba架构，数据分别走一遍实时，离线。看起来比较浪费，但有几个考虑：
1. Spark + S3 for batch processing
2. 会有补数据的需求，通过实时计算并不一定能满足，比如通过EventTime进行计算，并非Kafka中到服务端的时间
3. 不同的存储解决不同目的

Samza的问题：
1. 不能动态扩展
2. 部署较为不便