Kafka 2016 Summit上Uber工程师Danny Yuan分享了一个Streaming Processing PPT，如何解决Uber里Operation Team所需要的需求。看了整个视频觉得介绍很细致，这对于大部分LBS （Location Based Service）有很好的借鉴意义。
业务需求
Realtime OLAP
对于Operation部门而言，实时性很重要：
- 当前时间点，全球有多少量车在运行？有多少量车在空驶？
- 最近10分钟内，有多少UberX（类似于滴滴中的商务专车）在SF出现，热点地区在哪里？
- 每个区域的平均行驶时间、以及其他指标分别是多少？
作者给出了一个示意图，我们可以解读下：
- 右侧是一个湾区的地图，通过蜂窝状六边形把坐标划分若干区域，红色就代表车的密集程度
- 左侧是该区域在过去N分钟内各项指标的变化情况，例如平均的形式距离，接单率，平均客单价等
- 通过筛选时间段、指标（Metric)等，可以全方面了解运营状况
这个图表让我相当了之前用TreeMap来监控集群利用率的场景，如出一辙。
- 左侧通过HeatMap显示各个机架上的不同时间段上Metric变化情况
- 右侧则是各指标在时间段上分布的场景
只不过在机器运维的Portal上显示的是，只不过我们面对的是集群，Uber面对的是车与地图：）
CEP（Complex Event Processing）复杂事件处理
例子：
- 有多少个司机在最近10分钟内取消了3次接单以上？
- 如果发现后，会通过聊天软件与司机对话
Supply Position 供求关系可视化
在什么位置供大于求，什么位置求大于供：
- 黄色的点代表需求
- 蓝色的店代表供应
处理的挑战
如何表示车辆的位置数据
地理位置函数，一般用得比较多的是GeoHash，通过切分空间的方法把二维坐标，转化成一维的数字。两个区间的比较查询，就演变成一个一维的比较函数。
Uber的做法正好相反，将坐标转化成一个特定的区域，通过六边形的办法来逼近真实的位置。使用六边形有这样几个好处：
- 方便检索、查询、渲染
- 容易找到周围相邻的邻居
- 每个区域大小相同，形状相同
数据规模巨大
时间、空间、车辆状态、地理位置等组合会非常大
- 时间代表某一个时刻
- 空间在时间点上车的位置（例如LA，SF）
- 汽车的类型
- 状态（运行中，接单中，已接单出发地中等）
为了减少空间的规模：在地理位置、时间两个维度做了“取整”处理。通过六边形区域取整了地里位置，通过分钟级采样减少了其他状态，一天的数据量为
1dayofdata:300x10,000x7x1440x13=393billion
AI 代码解读
原始的数据为：
time, carID, locationX, locationY, status, .....
AI 代码解读
查询与计算的需求
- 车的种类、状态非常多、因此查询场景是面向多维数据的。
- 需要支持Heatmap，Top K， Histogram，count，avg，sum，percentage等计算函数
-
巨大数据量：
- 每秒百万级事件产生
- 每个事件中有20+Field
-
多种数据源
- 司机端事件
- 乘客端事件
Uber实时数据处理架构
分为5个部分：
- 日志、事件数据来源框架 - Kafka
- 数据清洗与处理，前置处理 - Samza
- 存储系统 - Elastic Search
- 数据读取，后置处理 - 自己开发的框架
- 查询与构建与查询 - 自己构建
- 应用层 - Web
数据采集与Kafka
这个Slides里面没有提到Uber架构，Google上找了一些相关的材料，整体架构如下：
数据来源有：
- Rider App
- Driver App
- API/Service（服务端）
- Dispatch （GPS 运行数据）
- Mapping & Logistic
日志、事件采集上在Kafka层包了Restful API，提供Java、Python、Go、NodeJS的SDK：
通过Samza进行清洗
主要有：
- Transformation（坐标转化）：GPS坐标是二维的，为了能够根据城市和地域查询，转化成更离散化的数据：ZipCode、Hexagon（六边形坐标）、城市等。 (Lat, Long) -> (zipcode, hexagon, S2)
- Pre-Aggregation：将一分钟数据归并成1分钟取整
- Join Multiple Stream：例如Driver Status、Rider Status进行合并
- Sessionization：将乘客的状态进行串联
From driver_canceled#window.time(10 min)
SELECT clientUUID, count(clientUUID) as cancelCount
GROUP BY clientUUID HAVING cancelCount > 3
INSERT INTO hipchat(room);
AI 代码解读
以上是一个ETL任务，每隔10分钟执行一次，既从Kafka中获得数据判断有问题的司机列表
通过这样的架构，支持运营人员能够在ES中清晰、索引的数据，获得实时分析能力：
同时由于在ES上层包装Query机制，也支持稍微复杂一些的离线查询。ES存储本身不是很好的离线方式，但对于离线查询频率不多的场景，也是够用的：
作者选型考虑
Lamdba vs Kappa
最终使用了Lamdba架构，数据分别走一遍实时，离线。看起来比较浪费，但有几个考虑：
- Spark + S3 for batch processing
- 会有补数据的需求，通过实时计算并不一定能满足，比如通过EventTime进行计算，并非Kafka中到服务端的时间
- 不同的存储解决不同目的
Samza的问题：
- 不能动态扩展
- 部署较为不便
一些看法
对于数据运营团队而言，重要的是实时性、另外就是大规模、准实时Muti-Dimension OLAP能力，特别是面对大量数据的场景下，如何在分钟级延时中获得筛选的数据与需求。这也是ELK这样技术方案受欢迎的原因。
对于地理位置类服务，数据预处理比较重要，例如根据IP获得省市范围，运营商等。根据GPS经纬度获得国家，城市，邮政编码等信息。根据地址信息获得坐标等。这些坐标转换有2类做法，在写入前通过Flume等插件计算，带来的问题是规模、并发不是很理想。第二种处理方法就是通过kafka这样管道读取，在下游进行计算与消费。在日志落到存储系统前，适当的清晰与Join是必要的。
没有一种存储引擎是万能，需要根据自己的需求来定制。ES提供的索引、列存储等能力还是非常适合对于事件类数据的存储与查询。目前最现实的做法还是数据收集一份，同时投递到多个系统中。
数据在写入存储前需要清洗，否则事后会带来非常大的代价。
