基于zookeeper和storm的车载流式计算框架

1、基于zookeeper和storm的车载流式计算框架摘要n 案例背景n 日本第一车载SAAS平台n 案例需求n 数据量剧增，吞吐量剧增n 车载机数量剧增+数据库连接数剧增n 任务分布不均导致实时性无法满足n 案例技术方案n 案例技术方案概览n 数据存储平台MongoDBn 实时计算平台Stormn 集群协调ZooKeepern 最佳实践n MongoDB介绍及最佳实践n Storm介绍及最佳实践n ZooKeeper介绍及最佳实践n 架构不足及经验分享案例背景1业务背景n 业务场景 物流公司 校车安全 油气公司 n 主要功能 运行数据记录 监视驾驶 违规警告 绩效考评 报表输出 危险地带 实时

2、监控日本市场占有率第一，积极拓展国内业务WebAPP中间件层硬件层通信层分析计算层1分析计算层2Web ServerWriteWriteWriteReadWritePollingReadRMDBPrimitiveLog基本Bs_WorkSheetBs_TimeChartBs_FreqDataBs_DigiData集計Master設定動態管理RMDBPrimitiveLog基本Bs_WorkSheetBs_TimeChartBs_FreqDataBs_DigiData集計Master設定動態管理RMDBPrimitiveLog基本Bs_WorkSheetBs_TimeChartBs_FreqDa

3、taBs_DigiData集計Master設定動態管理案例背景2技术背景市场竞争激烈，维持第一的保证PAAS一 数据量剧增+吞吐量剧增2013年车辆数：10000台数据量：3000w/天吞吐量：80M/S ()1000200040005000700010000050001000015000200220042006200820102012车辆数车辆数車両数：10000Commn ServerAnalysisServerStasticServer分時日月年bs_primitiveLogWRW-62,3403,722,15729,763,261654,722,397 7,856,6448,198bs

4、_WorkSheet-RWRW8,422504,8754,032,87488,704,7821,064,448,769bs_TimeChart-RWRW10,873600,8744,804,430105,600,1001,267,200,296bs_FreqData-WR45027,0964216,7846,480,54077,760,054bs_DegitachoData-W-1,01060,079480,10010,560,006126,720,190数据特性Heavy Write Heavy Read Past Useless2015年车辆数：100000台数据量：30000w/天吞吐量

5、：800M/S：车载机发送数据最高频率2次/秒，每次发送4KB数据。10000台车载机的峰值为10000*2*4KB =80000KB=80M/S ：从这个数据可以算出实际平均吞吐量是62340*4KB/60 = 4M/s, 但是固定时间点车载机会同时向云端发送运行数据()此表抽出通信中间件分析集计中间件DBbs_primitiveLogbs_WorkSheetbs_TimeChartbs_FreqDatabs_DegitachoData设置信息写原始日志表车载机数据二 连接数剧增+客户剧增DB1DB2DB3DBn 通信服务器与各个业务DB建立长连接 目前的SAAS平台已经有600租户 每个通

6、信服务器要与600个DB建立数据库连接归一化迫在眉睫NoSQL10W车辆时：1.需要17台通信服务器2.峰值：10W*8KB/S=800MB/S数据量长尾来袭 现在有600个客户公司，未来？三 分析计算程序任务不均匀+实时性不够DB1DB2DB3DBnAnalysis ProcessAnalysis ProcessAnalysis ProcessAnalysis Processn 现状 同一公司的车辆数据分布在同一数据库 分析计算进程与DB一对一地分析数据n 问题 若一个公司有很多辆车，那么一个分析计算进程应付不过来，无法分散计算 若一个公司车辆很少，那么该公司对应的分析进程将没有任务，浪费服

7、务器资源 因为分析进程的缓慢执行，车辆的实时运行数据无法反映到客户端案例技术方案概览案例技术方案概览通信集群消息队列+存储计算集群数据库集群协调集群应用集群接受连接负载均衡通信直连事件网络消息发布消息订阅数据库操作存储消息负载均衡-数据分片主从备份负载均衡无模式数据分解数据分析数据统计日志分析数据存储模式隔离私有仓库节点管理任务分配状态维护Open APIBrowseripadPhoneLibeventKafkaMongoDBStormSQL ServerZooKeeperIIS、WCF数据存储平台MongoDBMongoDB是一个介于关系数据库和非关系数据库之间的产品，是非关系数据库当中功能

8、最丰富，最像关系数据库的。他支持的数据结构非常松散，是类似json的bjson格式，因此可以存储比较复杂的数据类型。Mongo最大的特点是他支持的查询语言非常强大，其语法有点类似于面向对象的查询语言，几乎可以实现类似关系数据库单表查询的绝大部分功能，而且还支持对数据建立索引。它的特点是高性能、易部署、易使用，存储数据非常方便。主要功能特性有： 面向集合存储，易存储对象类型的数据 模式自由 支持动态查询 支持完全索引，包含内部对象 支持查询 支持复制和故障恢复 使用高效的二进制数据存储，包括大型对象（如视频等） 自动处理分片，以支持云计算层次的扩展性 支持RUBY，PYTHON，JAVA，C+，

9、PHP等多种语言。 文件存储格式为BSON（一种JSON的扩展） 可通过网络访问数据量剧增和连接数剧增的解决办法n 高频读写表抽出n 1公司：1DB模式N公司：1DBDB1DB2DB3DBnDB1DB1DB2DB3DBn选择MongoDB的原因n 增删改查操作最接近SQL，迁移容易，并可为其他业务使用n 支持数据的自动和手动分片，横向扩展容易n 性能测试结果接近预期，够用即可，不用追求完美n 支持数据复制、故障转移、横向扩展，基本符合RAS需求CAP定理性能关系不大2个mongos，3个分片，非安全插入：单独执行时：写线程10个并发时，每秒37000次写入读线程4个并发时，每秒43000次读取

10、同时执行时：写线程10个并发时，每秒达25000次写入读线程4个并发时，每秒达36000次读取：CPU 3.2GHz MEM 16G 千兆网卡MongoDB最佳实践n 逻辑部署和物理部署n 存储结构设计n 定期数据清除ShardsLoad BalanceConfig serverReplica setsmongosmongosmongosmongodmongodmongodmongodmongodmongodmongodmongodmongodmongodmongodmongodC 2mongodC1 mongodC3 mongodclientMongoDB的部署逻辑架构client192.1

11、68.0.2192.168.0.3192.168.0.4192.168.0.5192.168.0.6Shard1(master) Shard2(master)Shard3(master)Shard1(slave)Shard1(arbiter)Shard2(slave) Shard3(slave)Shard1(slave)Shard3(slave)Shard2(arbiter)Shard3(arbiter) Shard1(arbiter)Shard2(arbiter)Shard2(slave)Shard3 (arbiter)Config server1Config server2mongosmo

12、ngosmongosShard1：192.168.0.2,192.168.0.5,192.168.0.4,192.168.0.3,192.168.0.6(arbiter)Shard2：192.168.0.3,192.168.0.5,192.168.0.2,192.168.0.4,192.168.0.6(arbiter)Shard3：192.168.0.4,192.168.0.5,192.168.0.3,192.168.0.2,192.168.0.6(arbiter)Config Server1：192.168.0.2Config Server2：192.168.0.3Mongos：192.16

13、8.0.4，192.168.0.5，192.168.0.6MongoDB部署物理架构存储数据结构设计 “v”: “DTS19982221” “r”: “t” : 2013-11-14T08:17:00.016Z “d” : ”xxxxxxxxxxxxxxxxxx” “o” : “yyyyyyyyyyyyyyyyyy” , “t” : 2013-11-14T08:19:38.342Z“ “d” : ”xxxxxxxxxxxxxxxxxx” “o” : “yyyyyyyyyyyyyyyyyy” “s”: “zzzzz” 名字段要短 db.vehicle.save(“v”:” DTS19982221

14、 ”, “r: “t: 201399988772, “d”:”xxxxxxxxxxxxxxxx”,”o”:”yyyyyyyyyyy” db.vehicle.find(“v”:” DTS19982221 ”, “r: $elemMatch: “t:$gte: 201309898876251728定期数据清除n 清理程序（MonDC）运行在ZooKeeper集群上，自身无状态n 清理程序（MonDC）监视MongoDB内存使用量n 清理程序（MonDC）检测删除过期数据实时计算平台Storm与Hadoop类似，Storm为分布式实时计算提供了一组通用原语。可被用于“流处理”之中，实时处理消息并更新

15、数据库。这是管理队列及工作者集群的另一种方式。Storm也可被用于“连续计算”， 对数据流做连续查询，在计算时就将结果以流的形式输出给用户。它还可被用于“分布式RPC”，以并行的方式运行昂贵的运算。Storm的主工程师Nathan Marz表示：Storm可以方便地在一个计算机集群中编写与扩展复杂的实时计算，Storm之于实时处理，就好比Hadoop之于批处理。Storm保证每个消息都会得到处理，而且它很快在一个小集群中，每秒可以处理数以百万计的消息。更棒的是你可以使用任意编程语言来做开发。Storm的主要特点如下： 简单的编程模型 可以使用各种编程语言 容错性 水平扩展 可靠的消息处理 快速

16、Storm为什么号称实时计算系统？n 计算基于内存，较之磁盘有数量级之差n 使用消息队列做数据传输，速度比数据库快很多n 使用流式计算思想，数据可以源源不断的被处理n 任何一个节点都可以将中间计算结果反馈给用户为什么选择Storm？n 无需维护消息队列（Topic、LB、Broker），专注Producer&Consumern 满足业务需求，适合业务场景n 社区活跃，版本更新快n 性能卓越，易扩展Storm里的核心概念n Tuple&StreamTupleTupleStreamn Topology&Spout&BoltSpoutBoltBoltBoltBoltB

17、oltClusterStorm里的核心概念n Nimbus&Supervisor&Worker&TaskNimbusZooKeeperSupervisorZooKeeperZooKeeperSupervisorSupervisorSupervisorSupervisorWorkerTask:SpoutTask:BoltTask:Bolt提交代码，分配任务，监视Supervisor运行业务拓扑线程进程Storm里的核心概念n Stream GroupingSpoutBoltABoltBoltBBoltBoltCI know who am IIKnow:1whoamI:1I

18、knowI:1who:1am:1NOT PASS问题：如何保证tuple按照予定的路径发送到指定Bolt呢？Storm里的核心概念n Stream GroupingShuffle Grouping 随机分组， 随机派发stream里面的tuple.Fields Grouping 按字段分组，具有同样字段的tuple会被分到相同的Task。All Grouping 广播发送， 对于每一个tuple， 所有的Bolts都会收到。Global Grouping 全局分组， 这个tuple被分配到storm中的一个bolt的其中一个task。再具体一点就是分配给id值最低的那个task。Non Gro

19、uping 不分组， 这个分组的意思是说stream不关心到底谁会收到它的tuple。Direct Grouping 直接分组, 这是一种比较特别的分组方法，用这种分组意味着消息的发送者指定由消息接收者的哪个task处理这个消息。Storm里的核心概念n Stream GroupingBoltABoltBBoltABoltBBoltABoltBBoltABoltBAllFieldsShuffleGlobalStorm最佳实践n 计算拓扑设计n 数据顺序性保证n 基于统计的Worker和Task数值设置计算拓扑设计n 数据输入 FetchSpout MongoDB中读取车辆运行数据n 分析处理

20、AnalysisBolt 解码运行数据，如经纬度、违反等n 数据准备 DataPreBolt 为集计所需要的数据作准备，会与数据库交互n 集计处理 StasticBolt 统计一段时间或一天的运行数据n 数据库写 DBWBolt 所有涉及写数据库的操作由此BOLT完成计算拓扑设计MongoDBFetchSpoutAnalysisBoltAnalysisBoltAnalysisBoltDataPreBoltDataPreBoltDBWBoltDBWBoltCompany 1Company 2Company 3Company 4DBMLayerStasticBoltStasticBoltFetch

21、SpoutFetchSpoutAnalysisBoltAnalysisBoltSpoutDataCollBoltDataCollBoltStasticBoltStasticBoltFileds GroupingFileds GroupingFileds GroupingDTS0002, XXXXXXDTS0001, XXXXXXDTS0001, YYYYYDTS0002, YYYYYDTS0001, ZZZZZZZDTS0002, ZZZZZZZFields Grouping 按字段分组， 比如按VehicleId的模n结果来分组， 具有同样VehicleId的tuple会被分到相同的Bolt

22、s， 而不同的userid则会被分配到不同的Bolts。计算拓扑设计顺序性如何保证最大效率化基于统计的计算量均匀化最佳参数设置 通过反复的测试，得到一个接近能够处理完数据的比例设置，这里是2：10. 按照一个最佳数值设置建议：Worker个数是机器的整数倍，Task是Worker的整数倍。我们有7台机器作为Supervisor，所以Worker数设置为7，Task数值为14，于是得到FetchSpout个数为2，AnalysisBolt个数为12. 以上设置后，计算延时降低到1.3s左右Storm Cluster Number： 8Zookeeper简介 Zookeeper是Google的Ch

23、ubby一个开源实现，是高有效和可靠的协同工作系统。 Zookeeper能够用来leader选举，配置信息维护等，在一个分布式的环境中，需要一个Master实例或存储一些配置信息，确保文件写入的一致性等。 ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，包含一个简单的原语集，是Hadoop和Hbase的重要组件。 目前提供Java和C的接口。n 谁在用ZooKeepern ZooKeeper是什么lZooKeeper软件维护的是一个树形的数据结构l从任一个ZooKeeper节点看，树视图都一样/root /dataNode1 /dataNode2 /dataNode3 /d

24、ataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1lZooKeeper集群维护的是一个树形的数据结构/root /dataNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1/root /dataNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1/root /dat

25、aNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1/root /dataNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1/root /dataNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1写操作l

26、ZooKeeper各节点之间的数据结构是同步的/root /dataNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1Zookeeper核心概念数据结构Zookeeper核心概念节点类型节点特点功能PERSISTENT持久化目录节点，节点数据不会丢失保存配置文件，小量数据存储PERSISTENT_SEQUENTIAL顺序自动编号目录节点队列，共享锁等EPHEMERAL临时目录节点，客户端断开即消失作为客户端连接状态的监控节点EPHEMERAL_SEQUENTIAL临时自动编号目录节点同上，但可作为队列/root /da

27、taNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1n 统一命名服务（Name Service）n 配置管理（Configuration Management）n 集群管理（Group Membership）n 共享锁（Locks）n 队列管理 （Queue Management）Zookeeper的应用场景ZooKeeper的使用n 协调数据清理（ MonDC ）n 协调任务分配（TaskDistribute）n 协调Spout对MongoDB的数据访问n 集群节点监控数据清理程序（ MonDC ）开始MongoDB轮询清理警戒是否/shardProcTime /shard1：2013000912 /shard2：2013000523 /shard3：2013000943 /shard4：2013000125 /shard5：2013000229mongostat任务分配程序（TaskDistribute） spoutTasks