实时数据仓库相比较离线数仓，实时性更高，这就要求数据流尽量短，层次相对简化，相比较离线，这里的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就可以了。
整体的结构图
如图上所示，除了消费各自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，然后再跟右边的结果合并。一直到最后的节点
仔细观察，整个拓扑结构都是由一个个这样的最小单元组成的，那么最小单元是否可以继续抽象成通用的模型呢。拆分下这个单元，就是由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有一定的压力、拓扑结构比较多，运维比较麻烦
将多个数据流汇总成一个通用宽表，下游再有这主题的任何实时需求都可以很方便的使用和编写。提高工作效率，减少验证数据成本。
结论
方案可行
延迟在分钟内
除了运单等变化的字段其他字段可以与离线数据吻合
提高实时开发的效率
待续....困的受不鸟了。。。。
