【摘 要】为了满足移动终端游戏的精准运营需求，需要收集用户行为数据，实时分析业务状态，深度挖掘市场价值。基于Kafka、Flume、Storm、Redis等开源技术，构建了手机游戏大数据实时计算系统，实现了海量实时数据的采集、存储、计算和查询，并在现网实际系统中得到了成功应用。
【关键词】大数据 实时计算 开源 手机游戏
doi：10.3969/j.issn.1006-1010.2016.05.017 中图分类号：TP399 文献标识码：A 文章编号：1006-1010（2016）05-0079-06
引用格式：陈杰，苏洋，唐勇，等. 手机游戏大数据实时计算框架研究与实践[J]. 移动通信， 2016，40（5）： 79-84.
1 引言
在移动互联网时代，以手机游戏为代表的移动应用得到了飞速的发展。为了实现精细化运营，迫切要求针对海量数据能够实现实时计算结果以及秒级响应速度。以移动终端手机游戏平台为例，需要处理的流式数据主要包括手机游戏用户的PV/UV数值、页面浏览情况、手机游戏内容查找/登录/支付情况等，这些均要求实时数据的计算和分析，以便可以动态地获取用户访问数据，展示手机游戏平台实时流量的变化情况和用户行为习惯等。面对海量的业务数据量，传统的穷举所有可能条件的查询组合或者穷举条件组合的方法失效。
基于分布式处理机制和实时计算架构，将计算过程移至查询阶段，才能满足互联网业务海量数据计算和快速查询响应的需求。
2 实时计算处理流程
对互联网业务的海量数据（主要为日志流）的实时计算可划分为三大主要阶段：数据采集、实时计算处理分析和实时查询展示阶段。
在数据采集阶段，通常采用主要互联网公司提供的开源的海量数据采集工具，满足每秒数百MB的日志数据采集和传输要求，如Facebook的Scribe、LinkedIn的Kafka、Cloudera的Flume，淘宝的TimeTunnel、Hadoop的Chukwa等。
在数据实时计算分析阶段，首先将数据采集并存储在DBMS（Database Management System，数据库管理系统）中，然后通过查询和DBMS进行交互。但对于现阶段大量存在的实时数据，比如手机游戏交易/支付的数据，一般采用流计算技术。
在实时流计算框架方面，Yahoo推出的开源架构S4，Twitter使用的Storm，以及业界较为常见的Esper、Streambase、HStreaming等相关技术架构，均基于分布式并行计算（节点间的并行、节点内的并行）和热点数据的缓存处理等技术，提供实时计算服务。
在实时查询展示阶段，按照前端展示或者计算结果存储位置的不同可分为：1）直接提供数据读取服务，定期采用进程的内存镜像到磁盘或数据库全内存方法；2）采用Redis、Memcache、MongoDB、BerkeleyDB等内存数据库提供数据实时查询的半内存方法等。
3 手机游戏大数据实时计算
本文提出了手机游戏大数据实时计算架构，采用Kafka+Flume集群完成实时数据采集，利用Storm框架完成数据实时计算，采用Redis+HBase模式构建查询服务，满足了数据本地磁盘存储的安全性和长久性，实现基于内存提升查询速度。
3.1 实时计算架构
稳定可靠且高效的底层架构是实时计算的必要基础。图1给出了手机游戏大数据实时计算平台的总体框架，如图所示包含数据采集、数据存储、数据处理、数据应用四个层级。
数据采集对象主要是全体日志数据。按照统一规则整合，为数据应用提供实时数据。手机游戏日志主要包括两大来源，一是日志服务器集群实时上传的日志；二是业务接口后台服务器实时打印的日志。实时日志采集框架如图2所示：
将Flume部署在上述两大类日志源服务器上，作为海量日志实时采集的框架。如图2所示，Flume NG节点由Source、Channel、Sink三部分组成。Channel将Source和Sink连接起来。Flume的Source将外部数据源传递的数据封装成Flume数据模型的最小单位event。该数据源主要是业务接口实时产生的日志文件，使用操作系统原生类型exec来执行命令：tail-f/pathname/filename.xxx采集。此方式简单可靠，适合采集业务接口实时打印的日志。同时还需要打开Source的restart开关，当进程由于某种原因僵死后，可以自动重启。手机游戏行为数据由客户端统计SDK和业务接口产生。为了更好地区分有效日志与程序调试或其它用途的内容，约定了统一的实时日志前缀“[REALDATA]”来减少Flume Agent的日志量，提高了日志采集效率。
数据存储层提供了实时数据处理层需要的类分布式存储，主要采用了分布式消息队列（Apache Kafka）和Apache Zookeeper。通过Flume NG节点的Sink配置a2.sinks.k1.type=org.apache.flume.sink.KafkaSink，采集数据并实时存放在Kafka中。
数据实时处理层采用Storm实时从Kafka获取数据。在Storm中，首先需要设计一个用于实时计算的图状结构拓扑。拓扑将被提交给集群，由集群中的主控节点分发代码，将任务分配给工作节点执行。一个拓扑中包括Spout和Bolt两种角色，Spout发送消息，负责从Kafka中将数据流以tuple元组的形式发送出去；Bolt则负责转换这些数据流，完成计算、过滤等操作。Bolt自身也可以随机将数据发送给其他Bolt。由Spout发射出的tuple是不可变数组，对应着固定的键值对。相关的组件架构和数据流如图3所示：
Storm计算后的数据结果将被实时地写入Redis缓存、Mysql、HDFS、HBase中。其中写入Redis的数据用于实时展示运算结果，写入Mysql、HDFS中的数据用于后续的离线计算任务，写入HBase的数据可当作中间结果使用。
在实时数据应用层，将根据业务的需要来开发各类实时应用，比如手机游戏分发平台实时用户数、下载量、收入等。
3.2 实时计算方法
要实现大数据实时计算，仅有框架是不行的，还必须为每个业务设计一套专用的处理流程和算法。与离线计算相比，实时计算在算法上需要考虑的情况更复杂，这是因为实时计算所用到的存储资源远不如离线数据，但处理的时间限制要比离线计算严格，这都要求实时计算算法必须做更多的优化来提升计算的效率。
以下将以中国电信爱游戏基地手机游戏业务的海量计数需求为例，设计和优化相关实时计算算法。目前爱游戏手机游戏业务包含近千万的用户下载、手机游戏启动和付费数据，实时追踪这些数据是必须的，这也是个性化推荐、付费预测和用户画像等业务的前提。为此，急需设计一种算法可实时查看任意用户近24小时的下载次数、启动次数、付费次数等业务指标。
（1）简化方案
简化方案示意图如图4所示：
图中蓝色、黄色和绿色表示不同的用户，此方案为每个用户保存一份付费信息，它包含两个数据结构：
a）历史付费列表（简称“历史”），该列表中每个元素包含一个时间戳和一个整数，分别代表过去24小时中某一秒钟的启动数据并按时间排序。该列表最多可包含24×3600=86400个元素，但一般情况不会超过100个。
b）累计付费数据（简称“累计量”），代表截止到最后一次付费时的付费次数。
如图4所示，假设蓝色用户对应的数据是[（t1，a1）， （t2，a2）， …， （tn，an）]和A。这表示t1时刻的用户付费次数是a1，t2时刻是a2，以此类推，累计量是A。当用户付费数据不断进入消息队列时，处理进程（或者线程）P1， P2…会实时读取到这些信息并修改对应用户的数据。如P1读取到t时刻蓝色用户的付费记录时，会进行下面的操作：
a）得到当前时刻ct；
b）对数据库中蓝色用户加锁，加锁成功后读取数据，假设为[（t1，a1）， （t2，a2）， …， （tn，an）]，累计量为A；
c）累计量递增：A=A+1；
d）历史量更新：如果ct=tn，[（t1，a1）， （t2，a2）， …， （tn，an+1）]，否则更新为[（t1，a1）， （t2，a2）， …， （tn，an）， （ct，1）]；最后删除时间戳小于ct-24×3600的元素，删除的同时从累计量中减去对应时刻的启动量；
e）将新的历史和累计量输出至数据库并释放锁。
此方案可正确得到每个用户24小时内的付费次数，并且只要在资源（计算、存储和网络）充足的情况下，数据库中用户的付费次数是实时更新的。此方案也是分布式实时计算中最简单、最常见的一种。
（2）优化方案
简化方案中需要对数据库加锁，无论加锁的粒度有多细，都会影响计算效率。要想提高实时处理效率，减少锁是非常重要的。一种常见的做法是将并行操作串行化，MapReduce中的Reduce，将key相同的数据交给同一个Reducer处理。基于此原理将方案改造如图5所示：
新增一个key的HASH处理，保证相同的用户可以落到相同的处理程序上。由于不存在资源竞争，处理过程不需要对数据库加锁。此方案可大大提高计算效率，整个计算过程变为：
b）读取蓝色用户数据，设为[（t1，a1）， （t2，a2）， …， （tn，an）]，累计量为A；
c）累计量递增：A=A+1；
d）历史量更新：如果ct=tn， [（t1，a1）， （t2，a2）， …， （tn，an+1）]，否则更新为[（t1，a1）， （t2，a2）， …， （tn，an）， （ct，1）]；最后删除时间戳小于ct-24×3600的元素，删除的同时从累积量中减去对应时刻的启动量；
e）将新的历史和累计量直接输出至数据库。
步骤b）和e）省去了锁操作，整个系统的并发和吞吐量均大大提高。但此方案的缺点是存在单点隐患。一旦P1由于某些原因失效，则蓝色用户的数据将得不到及时处理，计算结果将无法实时保障。
在上述计算步骤b）和e）中总是把历史和累计量从数据库中读写。但在实际应用中只关心实时累计值，而历史数据仅用于计算累计值，并不需要实时持久化。为此，在最终方案中，区别对待历史和累计量，将历史和累计量都缓存在计算进程中，定期更新历史数据到数据库，而累计量则实时更新。最终改良优化的方案如图6所示：
计算过程为：
b）如果本地没有蓝色用户信息，从数据库中读取蓝色用户数据；否则直接使用本地缓存信息。假设为[（t1，a1）， （t2，a2）， …， （tn，an）]，累计量为A；
d）将新的累积量输出至数据库，如果满足一定条件（距离上次时间间隔足够远，或者积累的消息达到一定量），则将历史数据输出至数据库。
最终方案可大大降低数据库压力、I/O和序列化反序列化次数，从而提高效率。此方案对系统的稳定性要求更高，一旦P1失效则缓存数据将永久丢失。
（3）存储模糊化处理方案
在数据存储时，假设时间戳和整型均为long型（8字节），按照简化方案估算，每个用户所需要保存的历史数据大小为100×（8+8）=1600字节≈1.56kB，1000万用户的总量将有15G。若考虑到数据库和本地缓存，则系统存储量至少要30G，由此存在较大的空间浪费。 在优化过程中，为了降低历史的存储密度，将精度定义为小时级别，这样每个用户历史数据为24个，数据量为384字节，1000万用户的数据总量为3.6G，数据库和缓存的总存储量不超过8G。更近一步，使用一个字节保存当前小时段，使用包含24个long型的数组表示付费次数，则数据量将固定在196个字节，总存储量不超过4G。步骤如下：
a）得到当前时刻精确到小时的部分ct；
b）如果本地没有蓝色用户信息，从数据库中读取蓝色用户数据；否则直接使用本地缓存信息。假设为[time， a1， a2， …， a24]，累计量为A；
c）累计量递增：A=A+1；
d）历史量更新：令当前小时段为hour，如果ct=time，[time， a1， a2， …， ahour+1， …， a24]，否则更新为[ct， a1， a2， …， 1， …， a24]；最后将时间坐标小于ct-24的元素置为0，同时A减去对应时刻的付费数；
e）将新的累积量输出至数据库，如果满足一定条件（距离上次时间间隔足够远，或者积累的消息达到一定量），则将历史数据输出至数据库。
存储模糊化处理示意图如图7所示：
4 应用效果
4.1 实时业务监控
为了实时监控手机游戏平台的访问、下载、收入等情况，将实时统计好的数据包存放在Redis中，然后搭建web监控系统，图形化展示多维度的实时数据。同时针对重要的业务接口，监控每分钟的连接数等指标，及时发现异常日志并报警，可有效识别出攻击和恶意用户点击行为。经实际测算，每秒平均运算量超3000条，峰值超10 000条，日处理日志总数超3亿条。
4.2 用户实时关怀
为了提醒用户的消费行为，需要实时计算所有用户每一次的付费行为。如当用户的付费总额达到某一阈值，或首次在平台、渠道进行付费时，该系统会立刻给用户发送关怀短信，提醒用户的消费行为，或提醒其参加抽奖等营销活动。
该功能上线后，新增付费用户的注册转化率提升3.3%，手机游戏平台投诉用户下降明显，万投比下降8.2。
5 结束语
本文采用Kafka+Flume集群完成实时数据采集、采用Storm框架完成数据实时计算、采用Redis+HBase模式构建实时查询服务，构建了手机游戏大数据实时计算系统。在实时计算方面，为了满足千万级业务数据的实时查询需求，设计了两种算法。一是需要数据加锁的简化计算方案，二是为了提高效率，去除数据锁的并行操作分布式串行化处理的优化方案，同时还进行了数据分层处理优化，降低了数据库压力。在数据存储方面，结合模糊化处理方式优化数据存储，将存储资源要求减少到常规的12%。本项目成果在中国电信爱游戏基地平台得到了成功应用，每秒平均运算量超3000条，峰值超10 000条，日处理日志总数超3亿条。该框架可为其他业务平台大数据实时计算提供良好的参考和借鉴。
参考文献：
[1] 苏洋，刘晓军，唐勇，等. 游戏大数据平台研究与实践[J]. 电信科学， 2014（10）： 21-26.
[2] 周江，王伟平，孟丹，等. 面向大数据分析的分布式文件系统关键技术[J]. 计算机研究与发展， 2014（2）： 382-394.
[3] 胡宇舟，范滨，顾学道，等. 基于Storm的云计算在自动清分系统中的实时数据处理应用[J]. 计算机应用， 2014（S1）： 96-99.
[4] 唐云善，杨志. 一种高效的大数据实时性解决方案[J]. 计算机与数字工程， 2014（4）： 678-684.
[5] 赵辉，杨树强，陈志坤，等. 基于MapReduce模型的范围查询分析优化技术研究[J]. 计算机研究与发展， 2014（3）： 606-617.
[6] 张华，王东辉，吴@. 流式计算的分布式框架的应用[J]. 信息与电脑：理论版， 2014（10）： 142-143.
[7] 黄馥浩. 基于Storm的微博互动平台的设计与实现[D]. 广州： 中山大学， 2013.
[8] 高鹏. 当新媒体遇到“大数据”[J]. 广播与电视技术， 2012（10）： 38.
[9] 刘林林. 大数据时代数据分析与信息安全[J]. 网络安全技术与应用， 2013（12）： 59.
[10] 吕明育. Hadoop架构下数据挖掘与数据迁移系统的设计与实现[D]. 上海： 上海交通大学， 2013.
[11] 谢超. 大数据下的数据分析平台架构[J]. 程序员， 2011（8）： 55-58.
