SQL On Storm ——EasyCount系统概述

1. 摘要：

TDW很好的解决了海量数据离线处理问题，但是在如下场景下：实时报表，实时监控，实时推荐，实时分析，TDW无法满足需求。而storm是应对这些场景的利器，但是storm开发的门槛较高，对于大多数使用TDW的同学来说，若是能有一套支持storm的SQL，想必那是极好的。故此本宫，不，本团队开发了EasyCount以飧大众。

EasyCount使用SQL描述业务的实时计算的需求，并将SQL转化为基于storm的topology。相对于传统SQL，实时SQL面临诸多挑战，EasyCount通过不同的方法给出自己的解决方案。目前EasyCount系统已经上线运营，已经覆盖所有业务BG，接入了约30个业务，每天接入量约1000亿。

2. 背景：

针对海量数据处理，基于SQL的TDW取得了巨大成功，这足以证明SQL在数据处理领域的重要性。然而随着公司业务不断发展，对于数据越来越重视，如下这些实时数据处理场景的需求愈加迫切。

1、实时报表：显然老板们早已经不能满足于只能看昨天的业务收入情况，他们希望现在就看今天收入了多少。产品经理希望在新产品上线的1分钟以后，就知道当前的登陆人数。产品灰度测试出了问题，如果第二天才通过数据报表来发现，这是不能容忍的。所有这些指向的都是“实时报表”这样一类的需求。用SQL语言描述起来再简单不过：SELECT count(1) FROM productTbl。

2、实时监控：在公司内部已经有了各种各样的监控系统，并且多数都已经是成熟的稳定的运营的系统。但是这些系统通常都是专用的，或者针对某一类系统的监控，或者针对某一类指标的监控。重要的是，监控系统用到的数据多数是需要经过一次计算处理的数据，客观上，每一个指标监控后面跟着一个专用的计算子系统。这使得建立一个新的监控指标门槛是很高的。实时计算的任务就是以最灵活的方式把这些指标实时的计算出来，降低监控指标门槛。

SQL举例：

SELECT appid, sum(if(error, 1, 0)) FROM productTbl GROUP BY appid HAVING sum(if(error, 1, 0)) >100。

3、实时推荐：实时推荐系统是一个闭环系统，用户行为数据接入到实时计算平台，实时计算平台实时的计算各种指标，并存储起来（目前多数使用分布式KV系统）。实时推荐引擎会根据这些指标以及相应的推荐算法，向用户推送推荐信息。这个过程中，实时指标计算的越快，推荐的效果也就越好。这正是实时计算平台的重要的用武之地。SQL举例：

SELECT adid, count(qq) pv, countd(qq) uv FROM productTbl GROUP BY adid,

SELECT jobtype, count(qq) pv, countd(qq) uv FROM productTbl GROUP BY jobtype,

4、实时分析：通常实时分析和实时监控是差不多的，不过这里强调的实时分析相对于实时监控更加复杂的实时分析逻辑。举例来说，网站恶意流量数据的突增，这是一个实时监控需求，但是要分析这些恶意流量的路径是怎样的，这就是实时分析的范畴了。

3. 数据大系统

当前数据平台部数据系统平台系统架构如下：

TDBank负责数据的实时接入，目前已经成为数平数据接入的主要平台，覆盖公司各个BG，日接入数据量近万亿。并且数据接入是在秒级延迟内完成的。

TDW和TRC两大计算平台分别负责离线和实时计算任务，为公司各个业务和产品提供各种数据服务。

同时我们提供包括tPG，HBase，TDE等诸多平台为用户提供了各种场景的数据服务。

其中的TRC就是实时计算平台，EasyCount系统是TRC的重要组成部分，通过SQL来描述实时计算需求，并提交计算平台运行，本文重点介绍EasyCount系统的思想，原理和使用。

4. 数据与实时计算：

在EasyCount系统的设计和开发过程中，我们对于数据，实时这些概念有着不断深入的理解和认识，总结起来就是如下的四句话：

数据有模式

一切皆是表

计算要实时

分发多样化

4.1. 数据有模式：

尽管目前非结构化数据技术快速发展，nosql理念被越来越多的人所倡导。但是我们仍然认为，结构化数据是数据分析，数据挖掘领域的主流。 数据的价值在于其模式（schema）。不过在一般的传统数据库中，数据模式定义中的数据类型只能是简单类型，例如整数，浮点数，字符串，时间，最多会包含一个二进制的类型。这限制了结构化数据的定义，使得结构化数据表达范围很局限。不过通过扩展数据类型的定义，将map，array，struct这些复杂数据类型加入到数据模式的数据类型定义中去，结构化数据的表达能力便得到极大增强。

数据的模式可以理解为数据的特性（Feature），在实时计算中，时间这个特性是极为重要的，我们认为，任何一个（条）数据，都应该有一个属于自己的时间属性。

4.2. 一切皆是表：

既然所有的数据都有模式，那么很自然，所有的数据都可以被抽象为表（Table）。表不过是具有相同模式的一组数据的集合而已，这与其存储方式无关。因此，我们认为，存储在TDBank，TDW，HBase，TDE，TPG，MYSQL等等这些系统上的数据，都是具有模式的表。

在实时计算系统中表分为两类：流水表和关联表。通常流水表代表流式数据，是实时的，最新鲜的数据。关联表表示状态数据或者维表数据。

4.3. 计算要实时：

从对于数据的最顶层的需求来看，所有的离线计算都是一种妥协，理想情况下，任何一条新的数据的产生应该立即完成所有的计算，更新所有的状态。这才是一种极致的数据体验。不过显然是因为技术上的挑战以及实现上的成本和代价考虑，不会那样做。但是新的技术不断发展，那种极致的数据体验的梦想正在慢慢成为现实。

4.4. 分发多样化：

每个实时计算的业务都有着不同的用途，其对应的计算结果数据应该被存储在不同的系统中去。有的应该存DB，有的应该存KV，有的要入TDW，有的要送回TDBank。实时计算系统应该自己完成这所有的事情。

4.5. SQL是数据的语言：

同其他所有计算机语言相比，只有SQL是数据的语言，作为一个操作数据的人来说，使用SQL是最直接，最自然，最合理的表达方式，因此我们的EasyCount系统选择使用SQL作为描述语言。

5. EasyCount系统设计：

简单来说：

1、EasyCount提供定制的SQL语法（基于Hive-SQL的扩展）。

2、用户撰写自己业务EC脚本，包含SQL语句，以及表描述信息（SQL中用到的所有的表的描述）。

3、EC脚本被提交到EasyCount系统以后

1）生成抽象语法树AstTree

2）生成逻辑执行计划LPlan

3）生成物理执行计划PPlan

4）生成物理执行实例Topology（Storm-Topology）

5）提交到Storm-Yarn集群运行

4、不同于传统数据库的SQL任务式运行（有生命周期），EC-SQL是一个持续运行的实例，也就是说一旦提交，就会一直持续不断的运行，除非人为终止，否则不会停止。

5、EC-SQL是运行在Storm-Yarn之上的，这对于不同业务之间进行资源隔离，任务平衡调度非常有益。

6. EC-SQL语法：

如下是EC-SQL的语法定义：

sql : withquerys? insertquerys (withquerys insertquerys)*

withquerys : WITH subquery (, subquery)*

insertquerys : insertquery (, insertquery)*

subquery : (selectquery) alias

insertquery : INSERT INTO table selectquery

selectquery : SELECT expr (, expr)* FROM source [WHERE expr]? [GROUP BY expr (, expr)* coordinateInfo?]? [HAVING expr]?

source : table | alias | subquery | joinquery | unionquery

joinquery : [table | alias | subquery] LEFT JOIN table

unionquery : selectquery (UNION ALL selectquery)+

expr : column | function(column* [ACCU | SW]?) | foreach | execute

coordinateInfo : [COORDINATE BY expr]? [WITH AGGR INTERVAL Number SECONDS]? [WITH ACCU INTERVAL Number SECONDS]? [WITH SW INTERVAL Number SECONDS]?

Number : [0..9]+

foreach : FOREACH (var OF|IN expr GENERATE expr)

execute : EXECUTE (define executeBlock emit)

define : DEFINE var AS type DEFAULT val (, DEFINE var AS type DEFAULT val)*

executeBlock : { block (, block)* }

block : assign | ifblock | forblock

assign : (var | $var) := expr

ifblock : IF (expr block+)

forblock : FOR(expr block+)

emit : EMIT expr

关于EC-SQL语法，后续会有文章详细介绍。这里介绍几个关键点：

1、两大基本结构，with和insert。

在写sql的时候，如果业务逻辑过于复杂，通常要使用很多的子查询，这会使得SQL写的难以理解，很多时候还会产生重复计算等问题。EC-SQL为解决这一问题做了最大的努力。通过with语法，可以将子查询的作为临时表的形式表示，然后在其他的子查询中可以直接引用，这很像高级语言中的临时变量。With和insert语句可以任意组合，只需要保证前后顺序（变量定义一定要在变量引用之前）。通过这种支持可以将sql写的非常优雅。

2、关于时间分区。

在TDW上的数据表是按照时间分区存储的，TDW的离线计算任务也是针对分区进行的，数据的时间信息被隐藏在分区的定义中了，如果我要计算某张表每个小时的记录数，只需要按照分区分别计算即可。在实时计算中，没有了时间分区的概念，但是实际的计算还是要依赖数据时间的，举例来说，希望计算每5分钟的记录数目。考虑到时间属性对于实时数据的重要性，EC-SQL中对于数据时间提供了语法级别的支持。在进行聚合统计计算的时候，用户可以通过COORDINATE BY语法指定一个时间字段，通过WITH AGGR INTERVAL ?? SECONDS的语法指定聚合窗口的大小。这样在进行聚合函数统计，如count，sum等，的时候就实现了按照数据时间进行实时统计的功能了。另外在有聚合统计的子查询中，可以在SELECT子句中调用AGGRTIME关键字，表示当前聚合窗口的归一化时间（当前窗口的起始时间的秒数表示）。

3、函数支持（参考hive函数支持，更详细的函数介绍参考后续相应的文章）

7. 问题，挑战，解决方案

EC-SQL进行实时计算面临诸多问题和挑战，主要表现在：

l 去重统计（distinct）

l 连接运算（join）

l 滑动累加窗口

l 复杂数据类型

l 多平台数据分发

l 性能

l 稳定性

l 容灾

7.1. 去重统计

EasyCount是基于storm开发的，因此一些底层特性依赖于Storm平台提供的功能。MR框架在MAP和REDUCE之间进行shuffle的时候是提供排序功能的，也就是reduce收到的数据是经过排序的，这为如下多中场景的计算提供了可能：

1）聚合函数的去重计算；2）排序计算；3）TOP-N计算等。然而在Storm底层SPOUT和BOLT，以及BOLT和BOLT之间的shuffle是简单的hash计算，而没有任何排序功能。之所以这样是有原因的，Storm是流式计算平台，处理数据的基本单元是记录，因此在平台本身不存在排序这样的语义。这带来一个问题，如果我想计算每1分钟内的qq的独立登录数的话，我的处理单元是1分钟的数据，需要对qq号进行去重，显然无法直接利用底层shuffle功能了。

解决上述问题的一个简单做法如下，首先通过group by对qq号进行shuffle，在进行一次count计算即可。具体sql如下。

With (Select AGGRTIME agtime, qq from src group by qq coordinate by unix_timestamp()*1000 with aggr interval 60 seconds) tmp \

Insert into dest Select count(1) from tmp group by 1 coordinate by agtime with aggr interval 60 seconds

这种方法可以实现精确的uv统计，不过也有弊端：

1）这条sql在实际执行过程中，会将1分钟内所有的qq号码缓存在内存中的，并且需要一次全量的从spout到bolt的消息传输，增大了系统资源占用。

2）这条sql最终被翻译成spout+bolt+bolt的结构，有些过于复杂。

我们在实际的业务需求接入过程中发现：如上述这类uv统计需要的大多数情况下，并不要求绝对的精确统计，这其实也很容易理解。如果我计算出来一款产品某1分钟的uv是100，但是精确的数字是101，那么这一点点的差别，在任何实际应用中都是无关紧要的。这提示我们，如果通过牺牲一定的精确性，能够降低资源占用，提高系统运行效率，这是一件绝对利大于弊的事情。因此，针对uv统计这样的场景，我们开发了专用uv统计的聚合函数countd，可以实现误差率在千分之5以内的uv统计。该函数主要的原理是基于HyperLogLogPlus算法进行实现的，本文不做具体介绍，可以参考相关论文。

那么使用countd如何实现上述的需求呢？如下面sql所示：

Insert into dest select countd(qq) from src group by 1 coordinate by unix_timestamp()*1000 with aggr interval 60 seconds

更多的用例参考后续文章。

7.2. 连接计算

Join在标准sql中是非常重要的语法，在传统的数据库和数据仓库的设计模式中，经常通过主键的方式把数据组织成为星形模式或者星座模式等。而进行数据计算的时候，则需要通过join来把各个数据表连接起来。

在实时计算中同样存在类似的问题，例如实时数据中只包含ip字段，但是在进行实时统计计算的时候，我们需要按照国家或者地区，以及其他属性进行统计。这些ip对应的属性信息需要通过ip_table这样一张表进行关联得到。这就是join的需求。

基于MR的Hive通过将多张数据表的关联字段（ip）作为key进行排序，然后在对每组key对应的多个表数据通过循环的方式把数据关联起来并且输出。这是是一种离线思维的计算模式，对于实时计算来说是没有办法这样做的。

EasyCount是这样解决关联的问题的，如前面所述，表分为流水表和关联表，EasyCount中只允许流水表对关联表进行左外关联，关联表一般存储在kv系统或者db中。实际的计算方法是，每来一条新的流水记录，就用这条记录中的key到关联表中查询，然后将查询到的数据和流水记录合并然后输出到下一级计算模块。举例如下：

Insert into dest select country_id, count(qq) from src left join dim on src.ip=dim.ip group by dim.country_id coordinate by dtTime with aggr interval 60 seconds

其中的src是流水表，而dim是存储在kv系统中的关联维表。

还有一类需求，是要将两张或者多张流水表进行关联，一个常用的场景就是计费数据实时对账功能。有两张流水表，分别是用户付费流水日志，和计费服务器db的binlog日志流水。对于任何一个用户的付费行为都会在这两张表中各产生一条记录。如果两张表的数据记录不能对上，则可能意味着一些危险行为，比如是否有人直接操作db等。传统的做法是将两张数据表按照qq号码进行关联，然后进行对账，但是这通常只能离线的进行，在实时对账的场景下，这种join运算是很难完成的。对此，EasyCout提供如下的解决方案：

1）将两张数据表适用union的方法合并到一起，并分别打标

2）按照1分钟的粒度按照用户qq号分组进行聚合计算，算出分别在两张表中的计费结果，相减得出一个差值。通常情况下，这个差值应该为0，但是因为时间差的缘故，即使没有出错的情况下，这个值也有可能不为0。在这种情况下，可以通过一张kv关联表，以qq为key对这些差值进行累加，如果类加到一定的时间以后仍然不为0，那就可以进行对账告警了。

对于这类需求还有一种更加直接和彻底的方法，那就是将其中的一张表实时的写入到TDE中去，另外一张表通过少许延迟然后与TDE中的表进行关联。这种方法对于外部维表资源占用较大，在实际的应用中需要权衡处理。

7.3. 滑动累加窗口

实时计算和离线计算本没有本质区别，不过是计算的时间粒度越小而已。然而随着时间粒度不断变小，衍生出一些新的需求，主要表现在滑动窗口和累加窗口的计算。

一个普通的需求场景是，计算每1分钟的统计指标（例如登录数），但是有的时候需要计算在当前小时截止到当前分钟的登录数，这就是累加窗口统计。另外一个场景是，每分钟计算过去5分钟的登录数目，这就是滑动窗口统计。这两类需求在实时报表，实时监控，实时分析等，都有着特定的作用。

针对这样的需求，我们对滑动和累加窗口统计提供了语法级别的支持。

举例：统计以下3个指标，每分钟的登录数，当前小时截止到当前分钟的登录数，过去5分钟的登录数。

Insert into dest select count(qq), count(qq ACCU), count(qq SW) from src group by 1 coordinate by unix_timestamp()*1000 with aggr interval 60 seconds with accu interval 3600 seconds with sw interval 300 seconds

7.4. 复杂数据类型

如前所述，为了完善结构化数据的表达能力，加入复杂数据类型是很重要的，EasyCount特别增加了对于复杂数据类型map，array（list），struct以及二进制数据类型binary的支持。这样在EC-SQL中的表达式计算中可以进行更加复杂的计算。

举例来说，如果希望计算每个人每5分钟内的新增好友列表，可以使用如下的方法。

Insert into dest select qq, collect_set(qq_friend) from dest group by qq coordinate by dttime with aggr interval 300 seconds

其中qq和qq_friend是原始数据中的qq及其好友。Collect_set函数是一个聚合函数，就像count的功能是计算一组数据之和一样，collect_set计算一组数据的集合。在数据存储的时候通过不同的分隔符进行序列化。

由于复杂数据类型的引入，对于复杂数据类型的处理则需要额外增加大量的函数，后续会有专门文章介绍。

有了复杂数据类型以后，虽然增加大量相关函数解决类型操作问题，但是仍然不能满足由于类型多样化而带来的更加复杂的类型操作需求。举例来说有了array类型以后，就希望能够有一种方法可以对array中的每个元素进行过滤，或者一些复杂的处理需求，比如累加，或者带条件判断的处理等。

基于这样的情况，我们开发了foreach和execute两个表达式操作功能。

7.5. Foreach语法

Foreach语法实现如下的功能，对于array类型或者map类型中的每个元素，对其进行处理，然后生成一个新的元素，将所有新生成的元素组成一个新的array数据类型。

举例：用户的记录中包含一个点击流，格式如下：qq|path, 其中path格式a,b,c,d,e,f...。表示一个qq号码对应的点击流，现在需要计算出所有的前驱后继三元组：{a,b,c},{b,c,d},{c,d,e},{d,e,f}等。通过foreach语法可以实现这样的功能。

select qq,

FOREACH(

idx of path

GENERATE

if(

idx<size(path)-2,

array(

array_get(path, idx),

array_get(path, idx+1),

array_get(path, idx+2)

),

null

)

) from tbl

其中的path就是点击流路径array类型，idx是array的下标，上面语句的含义是，针对path中的每个元素，生成一个三元组，然后将这些三元组组合为一个新的array类型输出。

对于输入为qq|a,b,c,d,e,f的记录，输出的记录表示如下：

qq|a;b;c,b;c;d,c;d;e,d,e,f（|是一级分隔符，逗号是二级分隔符，分号是三级分隔符）

7.6. Execute语法

Foreach语法解决了很多问题，但是仍然有很大的限制，为了提供更加灵活的计算语义，我们提供了execute语法。

Execute语法的思想是：在一个execute表达式内部，可以定义若干局部变量，然后通过循环或者条件判断等方式，对这些局部变量进行任意的赋值。在所有的赋值操作完成以后，通过emit子句计算希望输出的结果表达式。理论上说，通过execute语法可以解决任何的计算问题。

举例：用户记录中包含了用户最近几次游戏得分情况，我们希望计算用户得分的平均值。数据格式如下：qq|records，其中records是array类型，格式x1,x2,x3...。使用execute语法实现这个功能的代码如下：

Select qq,

EXECUTE(

DEFINE cnt as int default 0, summ as bigint default 0, idx as int default 0

{

FOR($idx<size(records)

$cnt := $cnt+1,

$summ := $summ+array_get(records, $idx)

)

}

EMIT $summ/$cnt

)

From tbl

7.7. 多平台数据分发

EasyCount是一个实时计算平台，本身并不包含存储资源。那么计算的结果数据必须存放在第三方平台上。在实际应用中不同场景需要将数据存储到不同的平台上去，一般来说，实时报表类计算结果需要存储到db，一般是mysql或者tpg。用于客户查询的数据需要存储在hbase。用于实时推荐类数据需要存储在kv系统，如TDE，CKV等。还有很多实时分发的数据需要写回TDBank。

EasyCount可以适配所有这些系统，存储在这些系统上的数据都被抽象为表，只不过是表的类型不同而已。总体上根据平台不同，EasyCount目前支持如下几种表类型：tube（tdbank流水表），mysql，tpg，hbase，tde，redis，ckv（即将支持）等。

同时EasyCount表的输入输出方式可以有以下几种：

1）流水输入：tube

2）关联输入：tde，hbase，ckv，db（mysql，tpg）

3）输出：所有类型表。

7.8. 性能

衡量EasyCount系统的性能主要从两个指标进行考虑：吞吐量，实时性。

7.8.1. 吞吐量

Storm官方文档中给出的吞吐量的测试结果是，单流水线处理能力约为20000条/s。这样的测试结果对于我们一些大的流水表来说性能是不够的。为了提高吞吐量，EasyCount系统在处理spout和bolt之间的数据传输时，采用了打包和压缩的技术，加大每条传输结果的大小，最大单条消息为4KB。经过实际现网的测试情况来看，这能够大大的提高传输吞吐率。

7.8.2. 实时性

EasyCount根据不同的业务场景提供的服务的实时性级别也是不同的。

对于实时更新类应用，每来一条数据都是实时更新外部数据库。实时性可以保证秒级延迟（虽然Storm本身提供毫秒级别延迟的服务，但是EasyCount目前没有这个承诺，原因在下一节进行分析）。

对于统计报表类需求EasyCount提供分钟级别的实时性服务。通常按照分钟进行数据统计的时候，系统默认有最少30秒的延迟用来进行超时检查，以确保最大程度的数据计算准确性。如果有更高的实时性要求，超时时间可以设置。

7.8.3. 吞吐量vs实时性

在追求实时性和吞吐量两个指标的时候，这里面是有一个权衡的，通过打包来提高吞吐量的同时，也会因为打包导致数据的延迟。这是不可避免的，因此在实际的业务中，需要对两者进行权衡。

从目前EasyCount接入大大多数需求来看，通常对延时的要求并没有那么高，秒级延迟已经是足够了，因此EasyCount系统在设计上更加侧重于提高系统的吞吐量。

这里有一个疑问，为何实时计算系统的实时性的要求却没有那么高？事实上，实时计算系统并非在线查询系统，实时计算系统的计算结果对于线上业务的影响都是异步的。可以想象，在一个实时推荐的系统中，当用户在前台点击一个商品时，这样的一个事件会在秒级时间内发送到实时计算系统，触发新的指标的状态更新，这立即影响到我们下一次为用户推荐的商品的选择。但是，以目前的技术框架，实现代价来看，无法实现在用户点击鼠标和松开鼠标的一瞬间完成，数据传输，指标计算更新，影响推荐算法，然后立即就返回受到这条记录影响的推荐结果。但是用户第二次点击鼠标的时候，那可能至少是几秒钟以后的事情了。因此只要能在用户第二次点击之前完成实时计算即可了，因此，这里追求实时性还不如追求吞吐量来的实惠。

7.9. 稳定性

系统稳定性主要表现在以下两个方面：资源不足以及外部系统影响的场景。

系统资源不足：如果一个业务计算消耗较大，但是实际提供的资源却不足的情况下，极有可能出现内存溢出，系统崩溃等问题。对于数据流量过大而处理吃力的情况，EasyCount通常会触发流量保护机制，使得数据处理变慢，进而触发相关告警机制。但是在内存占用较大的场景，仍然无法避免内存资源不够导致的内存溢出问题，因此在业务开发时，仔细的资源评估是非常重要的。

外部系统影响：通常如果因为外部系统故障，例如tde查询失败，hbase写入失败等问题，一般根据实际情况可以配置不同的处理策略，例如tde查询失败，会进行重试，hbase写入失败可能会重试若干次以后丢弃处理，等等。

7.10. 容灾

容灾在任何系统中都是一个特别重要的课题，其主要指标是在系统中的节点出现故障的时候，系统的容错表现。Storm平台本身提供了ack机制，应用程序可以通过ack机制实现分布式事务以实现容灾处理。但是从实践经验来看，基于ack的事务机制的容灾实现设计过重，其带来的性能损失太大，对于我们目前动辄千亿级别的数据处理来说不可接受。

因此EasyCount系统在容灾性和吞吐量上选在了吞吐量，其设计本身都是处处考虑系统的吞吐性能，放弃了完善的容灾特性的追求。

EasyCount对待容灾是这样看的：

1）实时应用场景中对于容灾的需求有多大？从我们已有的多数案例来看，通常应用对于系统少量数据丢失是可以容忍的。另外就是，如果数据已经产生了丢失，很多时候，通过任何手段补充回来已经失去了时间效力，实践中对于这些丢失的数据可以不处理。

2）确实有一些对于数据的容灾性有较为苛刻的要求的案例，比如实时数据对账，实时监控等场景。EasyCount给出的解决方案如下：多实例并行运行，同一个业务逻辑启动两套程序。一定有人会觉得这样太多浪费，但是事实上，像hadoop这样的系统为了保证数据容灾，本身数据默认都是有3个备份的，因此如果真的是非常严苛的计算场景，两套计算逻辑也不算是非常浪费的。另外的一个容灾方案是，数据回退，我们基于tube这样的消息系统进行计算，如果数据计算过程中系统故障，为了最大程度的保证数据完整，我们可以通过回调tube的offset的方式，将过去一段时间的数据重新计算一下。

综上所述，EasyCount本身对于容灾确实是较弱的，在有节点计算失败的情况下，确实会丢失一些数据，然而，Storm系统本身具有work进程自动调度的机制，这也就是说即使部分节点失败，但是很快就会恢复。数据丢失只会局限在很短的时间窗口内。

8. 如何接入

EasyCount已经正式上线运营，如果你想进行接入申请，或者验证测试，按照以下几个步骤进行即可：

1）流水表数据通过TDBank接入

2）准备好业务所需输出表，关联表等（例如输出为mysql，则需准备好相应的建表和db链接参数，用户名密码等）

3）撰写业务脚本，使用IDE进行编译调试

4）通过我们的trc前台上传脚本

5）在trc前台新建任务

6）等待管理员审核提交即可运行了。（由于实时业务对于资源占用比较敏感，因此需要管理员进行业务资源确认才能提交）

8.1. IDE使用

目前我们的IDE正在开发过程中，通常撰写一个脚本需要按照以下几个步骤完成：

1、配置相关输入输出表参数

2、撰写业务SQL逻辑

3、编译调试，生成ec脚本

下图是我们IDE的主界面。

8.2. 前台界面

用户通过IDE撰写完成脚本以后，需要在TRC的前台界面进行脚本提交和新建任务。主要有以下几个步骤：

上传脚本

新建任务

任务启动（由管理员完成）

8.3. 如何撰写EasyCount脚本

EasyCount脚本是一个以.ec为后缀的文本文件，主要包含两个部分：sql逻辑和表描述。（可以通过IDE生成，也可以直接撰写）

SQL逻辑是实时业务的sql，是主要的计算逻辑。表描述是对在SQL中用到的所有表的描述，主要包括表类型，表字段定义，表的数据分隔符等。针对不同的表还需描述相关的属性，例如mysql表需要定义其连接参数，主机ip，db名称，用户名，密码。

举例如下：teg_acct_test.ec

[system]

sql=INSERT INTO dest SELECT appId, count(qq), count(qq ACCU), count(qq SW) FROM src GROUP BY appId \

COORDINATE BY dTime WITH AGGR INTERVAL 60 SECONDS WITH ACCU INTERVAL 3600 SECONDS WITH SW INTERVAL 1800 SECONDS

[tabledesc-1]

table.name=src

table.fields=appId,string,:qq,string,:dTime,bigint,

table.tube.master=xx.xx.xx.xx

table.tube.port=8069

table.topic=ieg_bd

table.interfaceId=bd_download

table.field.splitter=;

[tabledesc-2]

table.name=dest

table.fields=appId,string,:cnt1m,bigint,:cnt_accu,bigint,:cnt_sw,bigint,

table.type=tpg

table.tpg.db.host=xxxxxxxxx-distribute.com

table.tpg.db.port=5432

table.tpg.db.name=xxxxx

table.tpg.db.username=xxxxxx

table.tpg.db.passwd=xxxxxx

table.field.splitter=;