好未来基于DorisDB的全新实时数仓实践
好未来（NYSE：TAL）是一家以智慧教育和开放平台为主体，以素质教育和课外辅导为载体，在全球范围内服务公办教育，助力民办教育，探索未来教育新模式的科技教育公司。截至2020年11月底，好未来在102个城市建立起990个教学点，业务范围覆盖全国331个地级市以及海外20多个国家和地区。
随着业务的发展，实时数据的分析需求日益增多，尤其在营销推荐、归因分析、业务辅助决策等场景下，实时数据分析所带来的效益提升是离线数据所不能比拟的。在这些业务场景的驱动下，好未来选择了DorisDB来支撑实时数据的分析应用。实现了数据秒级查询响应能力，构建了一个统一&快速&高效&灵活的实时数仓。
业务背景
业务场景分类
在教育场景下，根据数据时效性划分，数据分析处理可分为离线和实时两大部分：
离线
离线数据以8大数据域（日志、营销、交易、服务、教学、内容、学习、画像）建设为主，主要处理核心历史数据，解决“业务运营、分析师、算法”等海量数据多维度分析和挖掘等，采用批处理的方式定时计算。
实时
实时数据分析处理，主要包括由埋点产生的各种日志数据，数据量大，以结构化或半结构化类型为主；另外还包括由业务交易产生的业务数据，通常使用数据库的Binlog获取。
实时数据分析的需求越来越多，特别是在营销签单业务和在读学员是否续报等场景，需要实时数据来助力业务营销付费和续费目标达成。当目标没完成时，业务运营需要对数据进行多维度分析，找到原因，并快速做出决策调整等管理动作。
业务痛点
T+1的离线数据分析已经无法满足业务对时效性的需求，我们希望建设实时数仓来支持业务实时数据分析场景，解决如下痛点：
市场：想通过广告页投放策略，洞悉PV、UV等流量数据，如果出现异常，可快速分析和优化。但之前因为各种因素我们无法提供实时数据，对于业务来说T+1数据时效性滞后，参考价值有限。
销售：通过分析意向用户跟进和签单数据，根据当日销售目标，及时发现还有哪些管理动作需要优化。但目前是提供滞后数据，每日签多少单都通过人来统计，分析也是通过历史数据，分析效果很差。
在读学员续报：实时观测哪些学员续报了，老师需要做哪些续报动作。
课堂行为分析：分析课堂实时互动行为、答题行为等，阶段评测报告、课堂质量等。
算法模型：实时更新模型需要的特征数据，更准时的预测模型效果。
实时数仓目标
数据团队要提供灵活&丰富的分钟级的实时数据，并要保证数据的丰富性&准确性&及时性等。
丰富性
沿用离线数仓建模好的数据维度和指标，保证离线能用到的，实时也能用到。
准确性
实时指标的构建需要可以保证数据完整性和准确性。所有指标开发按照指标定义文档，线上使用DQC平台来监控数据准确性，实时发送异常数据等。
及时性
要保证数据的“新鲜”度，线上实时产生的业务数据和日志数据，要能及时地被用于数据分析，提升一线人员或业务的反馈速度。
实时数仓技术架构演进
2018年~2019年，基于Hive框架下的小时级任务方案；
2019年，基于Flink+Kudu的实时计算方案；
2020年至今，基于DorisDB的实时数仓技术架构。
基于Hive
在原有天级延迟的离线数据处理任务基础上，开发小时级延迟的数据处理链路，将核心数据按小时同步到Hive数仓中，每小时调度一次DAG任务，实现小时级任务计算。任务DAG示意图如下所示：
优点：
离线和小时级任务各自独立
代码逻辑复用性高，减少开发成本
可以使用离线数据覆盖小时级数据，进行数据修复
缺点：
小时级数据的延迟性还是很高，已无法满足业务对数据时效性的要求
MapRecude不适合分钟级频次的任务调度，主要是MapReduce任务启动慢，另外会过高的频次会产生很多小文件，影响HDFS的稳定性，以及SQL on Hadoop系统的查询速度
批量数据处理每次运行对资源要求高，尤其是当凌晨Hadoop资源紧张时，任务经常无法得到调度，延迟严重
基于Flink+Kudu
为了解决上面基于MapReduce小时级任务的问题，我们采用了流式处理系统Flink和支持增量更新的存储系统Kudu。
如上图所示，实时的日志数据通过Flume采集到Kafka，实时的业务数据通过canal实时同步数据库的binlog再转发到Kafka中，Flink再实时消费Kafka中的数据写入Kudu中。
在使用Flink+Kudu的实践中，我们遇到了如下几个问题：
Flink 基于 stream 语义，做复杂指标计算非常复杂，门槛高，开发效率不高，数据仓库更多使用批处理SQL
Kudu+Impala聚合查询效率不高，查询响应时间不能满足业务多维分析要求
使用Kudu需要依赖Impala、Hive等整个Hadoop组件，维护成本太高
Kudu社区不活跃，遇到问题很难找到相关解决方案，使用过程中遇到过宕机等各类疑难问题
基于DorisDB
基于上面方案的问题，我们开始对实时数仓进行调研，包括 DorisDB、ClickHouse、Kylin等系统，考虑到查询性能、社区发展、运维成本等多种因素，我们最后选择 DorisDB 作为我们的实时数仓，各系统的对比总结如下：
我们也深入考虑过ClickHouse，对于教育场景，一个学员要关联的数据维度多，包括课堂、服务、订单、教研等。在每个主题我们都会建设灵活且易用的星型数据模型。当业务想进行个性化自助分析时，仅需要关联相关表即可。但如果直接构建明细大宽表，随着业务不断调整，经常需要重构开发。这种情况下，ClickHouse的 join 能力弱，无法满足需求，而DorisDB强悍的Join能力，就成了我们应对业务变化的利器。而且 DorisDB支持CBO（基于成本统计的优化器），具备复杂查询的优化能力，从而可以快速的进行复杂实时微批处理任务，可以帮助我们更好的进行实时指标构建。
最终选择DorisDB的原因：
使用DorisDB可以让我们像开发离线Hive任务一样进行实时数仓的开发，避免了复杂的Flink stream语义，同时也能在功能上对齐离线指标，保证指标丰富性的基础上完成指标定义口径的一致，并且可以保证分钟级的数据可见性。
大宽表和星型模型的查询性能都很好，可以灵活高效的满足各类业务分析要求。
DorisDB 简单易用，运维管理成本低
基于DorisDB的实时数仓架构
系统搭建
整个系统，除了DorisDB集群之外，我们还搭建了下面两个配套系统
调度：使用Airflow，进行DAG任务调度
监控：使用grafana+prometheus，采集DorisDB信息并进行实时监控
实时数仓总体架构
基于DorisDB的实时数仓总体架构，主要包括下面三个部分：
数据源：业务数据（使用Flink实时同步mysql的binlog日志，写入到Kafka）、日志数据（包括H5小程序、APP、直播ipad客户端等埋点采集的各类日志数据，通过Flume写入到Kafka中）
数据存储：
采用 DorisDB的Routine Load直接消费Kafka中的日志和业务数据
使用DorisDB的Broker Load将Hadoop中的DWD、DWS、ADS等数据导入到DorisDB中
对于Flink等流式处理下系统，使用DorisDB的Stream Load方式实时将数据导入DorisDB
数据应用：
使用DataX可以将DorisDB数据导出到MySQL中
使用DorisDB的Export可以将DorisDB中的数据导出到HDFS中
DorisDB完全兼容Mysql协议，BI或业务系统可以使用Mysql Connector直接连接DorisDB进行使用
实时数仓数据处理流程
在实时数仓内部，也是按照传统离线数仓的方式，对数据处理进行分层处理：
ODS层，设置DorisDB的Routine Load间隔30秒消费一次Kafka数，写入到ODS表中
DWD层，按业务分析的需要建模DWD表，通过Airflow间隔5分钟，将ODS表中过去5分钟的增量数据写入到DWD表中
DWS层，对DWD表中的维度进行轻度或中度汇总，可以加快上层查询速度
BI层，通过自研的一个指标定义工具，分析人员可以快速的基于DWS构建报表，也可以衍生出一些复合指标进行二次加工。分析师也可以将取数口径中的SQL做临时修改，生成一个复杂跨主题查询SQL，来应对一些Adhoc需求场景。
DorisDB实时数仓具体应用
在好未来，为保证课堂上课数据、订单数据的实时分析要求，使用DorisDB支撑了课堂、订单等分析业务。下面以课堂、订单场景为例，从数据同步、数据加工等几个步骤拆解DorisDB在好未来应用场景的落地方案。
实时数据同步
在好未来，采用flink采集业务库的binlog数据，然后写入到kafka中，DorisDB只需要消费kafka对应的topic数据即可，整体流程如下图：
实时数仓数据处理
DorisDB内部的实时数据加工处理主要有如下操作：
缩短计算链路的长度，实时部分最多计算2层。dwd或dws层
增量计算，采用DorisDB的UNIQUE KEY模型，相当于(insert + update)，因此只计算增量部分即可
采用时间分区，多副本策略。既为了数据安全，又能避免锁表
离线表结构与实时表结构，保持一样，这样就可以用离线修复T + 1 数据
DAG任务调度
为了使DorisDB能在airflow上执行，我们封装了airflow调用DorisDB执行sql的算子，以便DorisDB的加工逻辑在airflow上被定时调度。
DorisDB任务执行状态的检查，由于不像T + 1，只需要判断昨天任务是否执行就行了，实时检查需要满足以下条件：
检查轮询间隔，需要根据不同的调度间隔，适当调整
检查轮询总时长，不能超过(调度间隔时长-10秒)
检查的范围，最小需要大于调度间隔，最大小于2倍的调度间隔
根据以上的实时调度检查条件，我们封装了基于DorisDB的实时调度的任务检查airflow算子，方便使用。
实时数据生产预警
为了监控DorisDB的实时数据生产情况，我们设置了三种预警：
1、检查DorisDB消费Kafka的任务，是否停掉了，如果停掉自动重启，重启3次依然失败，再发通知，人为干预
2、检查常规任务的执行，如果执行报错，就发通知。
3、检查数据源与DorisDB实时数仓ods层表，schema的对比，如果出现schema变更，就发通知人为干预。这样我们就能在白天实时了解schema的变更情况，不必要等到调度报错才发现，而且不影响线上数据产出。
DorisDB使用效果
提升业务收益
DorisDB在众多场景给业务带来了直接收益，尤其是DorisDB的实时数据与算法模型相结合的场景。比如教育的获客、转化、用户续报等业务，之前模型需要特征数据都是前一天的，所以模型也相对滞后。而我们通过大量数据分析得出结论：是当日行为和跟进数据，是最有价值的特征数据，这样模型效果较好。特别是意向用户识别模型，成为线索当天的历史积累数据的特征和前一天的历史积累数据的特征，分别训练模型后，线上实际预测效果相差2-3个百分点，AUC 0.752 和 AUC 0.721的差别，所以，当天的特征模型效果特别明显。
降低使用成本
用简单的SQL语义替代Stream语义完成实时数仓的开发，大大降低了开发的复杂度和时间成本，同时能够保证和离线指标的一致性。
结合使用宽表模型和星型模型，宽表和物化视图可以保证报表性能和并发能力，星型模型可以保证系统的查询灵活性，在一套系统中满足不同场景的分析需求。另外，明细表查询我们通过 DorisDB 外表的方式暴露查询，提升了查询的速度，大大降低了业务方的成本。DorisDB 的分布式Join能力非常强，原来一些需要查询多个 Index 在从内存中计算的逻辑可以直接下推到 DorisDB 中，降低了原有方案的复杂度，提升了查询服务的稳定性，加快了响应时间。
BI报表迁移成本低，我们前期BI可视化是基于 Mysql 构建的，某些看板不断优化和丰富需求后，加上多维度灵活条件筛选，每次加载超级慢，业务无法接受，当同样数据同步到DorisDB上后，我们仅需要修改数据源链接信息，SQL逻辑不用修改（这个超级爽，迁移成本超级低），查询性能直接提升 10 倍以上。
运维成本低，相对其他大数据组件来说，DorisDB只需要部署一种即可满足各类数据分析需求，不需要其他软件辅助，而且部署运维简单。
未来展望
DorisDB作为新一代MPP数据库的引领者，当前在多种场景下性能都非常优秀，帮助我们非常好的重构了实时数仓。目前DorisDB高效的支持了实时指标的计算，以及业务方在实时场景下的数据灵活探查和多维分析需求。DorisDB在集团内部各个业务线的应用越来越多，我们也将推动实时和离线数据分析进行统一，为业务分析提供更好的支撑。后继我们将分享更多DorisDB 的成功实践。最后，感谢鼎石科技的大力支持！