实时数据仓库-通用宽表

离线数据仓库都是T+1离线分析数据，运营人员今天看昨天的数据报表，当客户为店铺或者商品做活动并且想看当前的活动效果就只能等到明天来观察，然后根据结果调整活动策略，这样可能错过最佳的调整时机，当网站做大型活动，领导或公司就特别想看到活动实时带来的效益，比如阿里每年的双11，几分钟破亿、破10亿。当前最新的订单量、销售额等，随着大家对数据及时性的要求越来越高，实时计算应景产生。常见的开源实时计算框架有storm、s4、spark等。这里只讨论和关注storm部分。使用过storm开发过实时需求的同学都知道，storm对于单数据流的处理无论是开发难度或者处理执行效率都相当不错，部门有好多这样的任务跑的也一直比较稳定。但是当2个或2个以上的数据流进行关联，然后再进行逻辑处理，可能问题就凸显出来了，比如谁先到谁后到，谁一直不到，等待时间、过期怎么处理等等。今天这里主要讨论如何简单高效的将2个多于2个的数据流进行关联合并处理。如下图，我们要实现实时的通用宽表层。按照主题实现通用宽表，以后只要有关于这个主题的实时需求都可以从通用层计算实现。 Image

实时数据仓库相比较离线数仓，实时性更高，这就要求数据流尽量短，层次相对简化，相比较离线，这里的ods和明细表就可以合并等于从业务库实时同步数据写入到kafka中的一个topic。关于数据采集这里不做讨论，部门是部署脚本监控业务库的DML语句，将create、update、delete的数据实时同步写入到kafka。那么我们要实现多个数据流实时关联合并组成宽记录，
这里有几点要求
实时性
延迟要低，分钟级、秒级

一致性
保证数据的准确一致性，不能丢数据，宁可多算不可漏掉数据

易用性
实时数据有三种状态增、删、改，对于下游使用这个宽表要做到易用，能抽象通用的汇总模型

思路
1.如何解决谁先到谁后到谁迟迟不到的问题？storm的例子中有两个数据流的join的案例，是使用内存来存储先到的数据，设置过期时间，这样就有两个问题，过期就表明会丢数据，由于实时数据特点有增有改有删，即使关联上已经往下发送，但是考虑到update、delete，那内存中数据不能立即clear，内存消耗也是个问题，很容易吃满内存，溢出。如果放到内存不合适，那就借助存储数据库，比如hbase、redis，mysql等，这里第一版是使用的hbase，如果后期性能跟不上可升级为redis。2.如何解决更新、删除数据对之前合并结果的影响这里定义了一种操作叫做逆操作。比如金额最早为5元，后续变成3元，那这条记录我会把前后都传递给下游，3|5,下游如果有汇总，那就 - 5 + 3，如果只是合并不做计算，那就只取 3就可以了。
整体的结构图

就像倒立的二叉树，拓扑结构呈螺旋状，每个节点都是storm的bolt步骤，直到最后一个bolt就组成了我们想要的宽表A、B、C、D、E、F代表kafka的topicGHIK都是中间的bolt过程，都是hbase表，L代表最后的宽表，写到入kafka，等待下游消费使用。 Image
如图上所示，除了消费各自topic的数据写法不一致外，其他节点都是将左边和右边合并，然后继续往下发送，所以这些点都是可以抽象成一个通用的bolt来处理。
GeneralMerge
Map<String, String> left = (Map<String, String>)tuple.getValueByField("left");Map<String, String> right = (Map<String, String>)tuple.getValueByField("right");if(left != null && right!=null) { left.putAll(right);}获取左边右边，然后merge起来 图里边最小的单元如下图，都是2个流关联组成宽记录往下emit，然后再跟右边的结果合并。一直到最后的节点 Image

仔细观察，整个拓扑结构都是由一个个这样的最小单元组成的，那么最小单元是否可以继续抽象成通用的模型呢。拆分下这个单元，就是由2个数据流、关联关系、流到哪、数据流数据要存储到什么表中抽象出来就是左边的表，右表的表，表和表之间的关系【leftjoin、innerjoin】这些信息抽象成对象
Param leftParam = new Params(hb_table1,hb_table1_index,"L|R");Param rightParam = new Params(hb_table2,hb_table2_index,"L&R");

hb_table{1,2}代表数据存储的表名hb_table{1,2}_index代表数据自己的主键和另一方关联键的索引表，比如不都是主键之间进行joinL|R代表 左关联，代表左边的数据到了，会先去找右边的数据，即使右边的数据还未到，发现关联关系为L|R,左关联，左边也往下发送。最小单元的处理过程就是 先定义数据流的信息，读取topic，保存数据、索引表，找对方然后发送,如下代码
//获取数据流信息Map<String, String> leftMap = preParseProcess.process(tuple, leftParam); leftMap = save(tuple, leftMap); //保存emit(tuple,collector, leftMap); //找对方然后发送

数据传递
新增类型，封装到map中往下传递
更新类型，在更新的字段上标记新值、旧值，比如 new_value + split + old_value，下游使用时，已经告诉了你更新后的值、更新后的值，自己根据业务处理即可
删除类型，在每个字段后标记删除标签，如 old_value + split + DEL

优缺点
中间结果多、表多、操作hbase频繁对hbase有一定的压力、拓扑结构比较多，运维比较麻烦
将多个数据流汇总成一个通用宽表，下游再有这主题的任何实时需求都可以很方便的使用和编写。提高工作效率，减少验证数据成本。

结论
方案可行
延迟在分钟内
除了运单等变化的字段其他字段可以与离线数据吻合
提高实时开发的效率
待续....困的受不鸟了。。。。