本文选自《交易技术前沿》总第四十期文章（2020年9月）

肖钢、李宁、张海园 /中信建投证券信息技术部

摘要 ： 实时计算技术已经应用到广告、电商、游戏、文娱等各个领域，比如电商网站实时分析用户属性，基于分析结果给客户推送相关商品； 网络游戏实时分析玩家数据，进而对游戏参数和平衡性进行调整。 本文重点讲述中信建投证券利用实时计算技术搭建实时计算系统的过程和经验。

关键词：实时计算、实时数据处理、数据同步

1. 概述

1.1 什么是实时计算

实时计算是指对无限数据做无界数据处理，一般都是针对海量数据进行的，延迟要求为秒级。实时计算主要分为两部分:数据的实时入库和数据的实时计算。

1.2 应用领域

目前，实时计算技术已经应用到广告、电商、游戏、文娱等各个领域，比如电商网站实时分析用户属性，基于分析结果给客户推送相关商品；网络游戏实时分析玩家数据，进而对游戏参数和平衡性进行调整。

中信建投证券将其应用到金融领域，搭建了一套实时计算系统，实时汇总客户在股债基、理财、期权、两融、贵金属等业务的资产，为客户提供全方位的资产查询和分析服务。

1.3 实时计算系统简介

得益于逐渐成熟的实时计算技术，中信建投证券于2018年底建设了实时计算系统，为员工和客户提供实时资产查询服务，2019年又加入了实时盈亏、预期收益、理财资产可用日和可取日、计息天数等指标。

实时计算系统把客户在集中交易、两融、OTC、期权、黄金五大系统（以下简称“五大系统”）内的数据资产汇总到一起，并与五大系统的数据同步变化。

2. 建设经验

2.1 系统架构

五大系统包含的客户数量是千万级别的，如此多的客户，在开市期间的各种操作会产生海量数据，如何在如此大规模数据的条件下满足实时数据处理的需求，是实时计算系统必须要解决的问题。

开市前，实时计算系统从五大系统采集清算后的T-1日数据作为基础，开市后，实时计算系统采集五大系统变化的数据，加工后将其更新到基础数据里。为了便于理解，笔者在系统原有架构的基础上做了简化，详见图1：

图1 实时计算系统结构图

图1中实时数据流向如下：

1. Attunity 从五大系统数据源采集盘中产生的实时数据推送到Kafka；同时，行情转运工具从行情服务获取行情数据，推送到Redis；

2. 实时计算系统集群的Spark Streaming按先进先出的顺序消费Kafka内的数据；

3. Spark Streaming将实时数据加工后更新到Redis集群内的基础数据；

4. 其他系统访问致胜平台上的实时数据接口；

5. 实时数据接口从Redis集群内获取最新的客户数据，并将其转化为json格式的数据返回给其他系统。

系统建设过程中，项目团队还论证过另一种方案，即直接从五大系统数据库采集数据，汇总后提供给访问者。但经过讨论分析后，发现这种方案存在诸多缺陷：

1. 由于安全原因，五大系统不能直接暴露数据库给实时计算系统，这就需要五大系统开发专门的接口供实时计算系统使用，接口的开发和维护工作量很大；

2. 即使五大系统提供了查询接口，接口被调用时，五大系统也要从全部客户的数据里将被查询客户的数据检索出来，而且不止查询一个表，查询效率低下；

3. 此方案需要专门的服务器来运行，必须要投入一定数量服务器和网络资源，并不比其他方案需要投入的成本低。

2.2 关键技术

2.2.1 实时数据采集

系统建设初期，项目团队选择了Oracle GoldenGate（简称“OGG”）作为实时数据采集工具。

OGG软件是一种基于日志的结构化数据复制软件，能够实现大量交易数据的实时捕捉、变换和投递，实现源数据库与目标数据库的数据同步。Oracle GoldenGate通过解析源数据库在线日志或归档日志获得数据的增删改变化，再将这些变化应用到目标数据库，实现源数据库与目标数据库同步、双活。

一期系统建设完成后，项目团队发现OGG给运维团队造成了很大的工作量，每次五大系统数据库维护或升级，运维团队都要耗费大量时间来恢复OGG服务。因此，项目团队决定用一个更合适的数据同步工具–Attunity Replicate（简称Attunity）来替换OGG。

Attunity是一款基于数据库日志的异构数据库变化量同步工具。它采用B/S架构，通过浏览器图形化界面、鼠标点击拖拽的方式进行数据源、目标表的配置和维护，支持包括Oracle、SQLserver、DB2等十几种同构和异构数据库之间的跨平台数据同步。此外，它还支持并行工作，可以自定义并发度，这使得它比OGG延迟更低。

比于OGG，Attunity部署和维护更简便。Attunity无需在源端或目标端操作系统上安装软件，只需一台独立的服务器安装service，基于WEB的维护页面简单方便，只需要很少的操作即可完成数据同步配置，极大地减少了运维的难度和工作量。

2.2.2 中间数据存储

五大系统在开市期间产生海量的实时数据，即便是运行在集群上的Spark Streaming也无法在数据产生的瞬间将其完全消费，因此，就需要一个缓冲池来临时存放实时数据。综合各方面因素考虑，实时计算系统选用Kafka作为缓冲池。

Kafka是一种高吞吐量的分布式发布订阅消息系统。实时计算系统架设了专门的集群来运行Kafka，该集群有足够大的内存空间作为缓冲池存储实时流数据。Attunity 作为生产者，向Kafka实时流数据；Spark Streaming作为消费者。为了保障客户的每一步操作都能精确地同步到实时计算系统，系统严格按照先进先出的方式消费Kafka内的数据。

2.2.3 实时流计算

因为实时计算系统对时效性要求很高，所以必须有一个能快速处理数据的工具。项目团队选用Spark Streaming来承担数据处理的任务。Spark Streaming是Spark核心API的一个扩展，可以实现高吞吐量且具备容错机制的实时流数据处理功能，这种特性能满足实时计算系统低延时要求。

开市前，实时计算系统先将T-1资产推送到Redis里，开市后，Spark Streaming持续从Kafka里消费数据，然后对其进行处理。Spark Streaming内部的数据处理逻辑如图2：

图2 数据处理逻辑

图2中的数据处理层Spark Streaming与上下游工具配合，承担了实时流处理的工作。以客户开市期间的一笔股票交易为例，Spark Streaming处理该客户的资产过程简化后如图3：

图3股票交易同步逻辑

接口服务层负责将客户资产按照相关业务规则分类。为了给调用者提供一个大而全的资产汇总数据，系统在接口服务层把客户在五大系统里的资产数据汇总到一起，为每个客户生成了一条以客户号为唯一标识的json数据，返还给调用者。

接口返回的json数据内包含资产类型按币种划分包含：人民币，美元，港币。其中，有美元和港币资产的客户较少，业务相对简单；有人民币资产的客户较多，业务种类繁多。当上述各类资产单独存在于各自的系统时，资产和负债情况比较简单，无需繁琐计算即可得出其总资产、总负债、净资产等数据，但是如果要把五大系统的资产和负债汇总到一起，就要从全局的角度考虑各种情况，比如：客户在认购场外开放式基金时如何处理普通账户上的在途资产；客户用现金理财产品购买股票时如何修正其可用资金；客户提前赎回理财产品后如何更新其可用资金。为了解决这些复杂的问题，项目团队做了大量的调研和梳理工作，最终形成了一个包含五大系统资产的客户全景资产模型，图4是该模型的中客户人民币资产的内容：

图4客户人民币资产概览

2.2.4 对外服务接口

经过Spark Streaming加工汇总的客户数据，需要有一个容量大、响应快的容器来容纳；当需要对外提供数据时，还需要一个高并发的服务平台。中信建投证券自主研发的致胜平台 【1】 能够满足上述需求，项目团队基于致胜平台开发了一套接口服务。

致胜平台为实时计算系统分配了一组Redis集群，专门用来存储汇总后的客户数据。为了最大限度的提高Redis集群的吞吐量，数据在Redis集群内以序列化的形式存储。当其他系统调用系统接口时，接口从Redis里抽取相应数据并加工成json格式，返回给调用者。Redis里的数据保持实时更新，接口服务随时等待被调用，两者互不干涉。接口调用示意图见图5：