导读： 360商业化为助力业务团队更好推进商业化增加，实时数仓共阅历了三种形式的演进，别离是 Storm + Druid + MySQL 形式、Flink + Druid + TIDB 的形式 以及 Flink + Doris 的形式，依据 Apache Doris 的新一代架构的成功落地使得 360商业化团队完成了实时数仓在 OLAP 引擎上的一致，成功完成广泛实时场景下的秒级查询呼应。本文将为咱们进行详细介绍演进进程以及新一代实时数仓在广告业务场景中的详细落地实践。
作者｜360商业化数据团队窦和雨、王新新
360 公司致力于成为互联网和安全服务供给商，是互联网免费安全的倡导者，先后推出 360安全卫士、360手机卫士、360安全浏览器等安全产品以及 360导航、360搜索等用户产品。
360商业化依托 360产品庞大的用户掩盖才干和超强的用户粘性，经过专业数据处理和算法完成广告精准投进，助力数十万中小企业和 KA 企业完成价值增加。360商业化数据团队首要是对整个广告投进链路中所发生的数据进行计算处理，为产品运营团队供给战略调整的剖析数据，为算法团队供给模型练习的优化数据，为广告主供给广告投进的作用数据。
业务场景
在正式介绍 Apache Doris 在 360 商业化的运用之前，咱们先对广告业务中的典型运用场景进行扼要介绍：
-
实时大盘： 实时大盘场景是咱们对外出现数据的要害载体，需要从多个维度监控商业化大盘的目标状况，包括流量目标、消费目标、转化目标和变现目标，因而其对数据的精确性要求十分高（确保数据不丢不重），一起对数据的时效性和稳定性要求也很高，要求完成秒级推迟、分钟级数据康复。
-
广告账户的实时消费数据场景： 经过监控账户粒度下的多维度目标数据，及时发现账户的消费改变，便于产品团队依据实时消费状况推进运营团队对账户预算进行调整。在该场景下数据一旦出现问题，就或许导致账户预算的过错调整，从而影响广告的投进，这对公司和广告主将形成不可估量的损失，因而在该场景中，相同对数据精确性有很高的要求。现在在该场景下遇到的困难是如何在数据量比较大、查询穿插的粒度比较细的前提下，完成秒等级查询呼应。
-
AB 试验渠道： 在广告业务中，算法和战略同学会针对不同的场景进行试验，在该场景下，具有报表维度不固定、多种维度灵敏组合、数据剖析比较复杂、数据量较大等特色，这就需要可以在百万级 QPS 下确保数据写入存储引擎的功能，因而咱们需要针对业务场景进行特定的模型规划和处理上的优化，提高实时数据处理的功能以及数据查询剖析的功率，只有这样才干满意算法和战略同学对试验报表的查询剖析需求。
实时数仓演进
为提高各场景下数据服务的功率，助力相关业务团队更好推进商业化增加，截至现在实时数仓共阅历了三种形式的演进，别离是 Storm + Druid + MySQL 形式、Flink + Druid + TIDB 的形式 以及 Flink + Doris 的形式，本文将为咱们进行详细介绍实时数仓演进进程以及新一代实时数仓在广告业务场景中的详细落地。
第一代架构
该阶段的实时数仓是依据 Storm + Druid + MySQL 来构建的，Storm 为实时处理引擎，数据经 Storm 处理后，将数据写入 Druid ，运用 Druid 的预聚合才干对写入数据进行聚合。
架构痛点：
开端咱们试图依托该架构处理业务上一切的实时问题，经由 Druid 一致对外供给数据查询服务，但是在实际的落地进程中咱们发现 Druid 是无法满意某些分页查询和 Join 场景的，为处理该问题，咱们只能运用 MySQL 定时使命的方法将数据定时从 Druid 写入 MySQL 中（类似于将 MySQL 作为 Druid 的物化视图），再经过 Druid + MySQL 的形式对外供给服务。经过这种方法暂时可以满意某些场景需求，但随着业务规划的逐步扩大，当面临更大规划数据下的查询剖析需求时，该架构已难以为继，架构的缺陷也越发显着：
- 面临数据量的持续增加，数据仓库压力空前剧增，已无法满意实时数据的时效性要求。
- MySQL 的分库分表保护难度高、投入本钱大，且 MySQL 表之间的数据一致性无法确保。
第二代架构
依据第一套架构存在的问题，咱们进行了初次晋级，这次晋级的首要改变是将 Storm 替换成新的实时数据处理引擎 Flink ，Flink 相较于 Storm 不仅在许多语义和功能上进行了扩展，还对数据的一致性做了确保，这些特性使得报表的时效性大幅提高；其次咱们运用 TiDB 替换了 MySQL ，运用 TIDB 分布式的特性，必定程度上处理了 MySQL 分库分表难以保护的问题（TiDB 在必定程度上比 MySQL 可以承载更大数据量，可以拆分更少表）。在晋级完成后，咱们依照不同业务场景的需求，将 Flink 处理完的数据别离写入 Druid 和 TiDB ，由 Druid 和 TIDB 对外供给数据查询服务。
架构痛点：
尽管该阶段的实时数仓架构有用提高了数据的时效性、下降了 MySQL 分库分表保护的难度，但在一段时刻的运用之后又露出出了新的问题，也迫使咱们进行了第二次晋级：
- Flink + TIDB 无法完成端到端的一致性，原因是当其面临大规划的数据时，敞开业务将对 TiDB 写入功能形成很大的影响，该场景下 TiDB 的业务形同虚设，爱莫能助。
- Druid 不支持规范 SQL ，运用有必定的门槛，相关团队运用数据时十分不便，这也直接导致了工作功率的下降。
- 保护本钱较高，需要保护两套引擎和两套查询逻辑，极大增加了保护和开发本钱的投入。
新一代实时数仓架构
第二次晋级咱们引入 Apache Doris 结合 Flink 构建了新一代实时数仓架构，借鉴离线数仓分层理念对实时数仓进行分层构建，并一致 Apache Doris 作为数仓 OLAP 引擎，由 Doris 一致对外供给服务。
咱们的数据首要源自于维表物料数据和业务打点日志。维表物料数据会定时全量同步到 Redis 或许 Aerospike （类似于 Redis 的 KV 存储）中，经过 Binlog 变更进行增量同步。业务数据由各个团队将日志收集到 Kafka，内部称为 ODS 原始数据（ODS 原始数据不做任何处理），咱们对 ODS 层的数据进行归一化处理，包括字段命名、字段类型等，并对一些无效字段进行删减，并依据业务场景拆分生成 DWD 层数据，DWD 层的数据经过业务逻辑加工以及关联 Redis 中维表数据或许多流 Join，终究生成面向详细业务的大宽表（即 DWT 层数据），咱们将 DWT 层数据经过聚合、经由 Stream Load 写入 Doris 中，由 Doris 对外供给数据查询服务。在离线数仓部分，相同也有一些场景需要每日将加工完的 DWS 数据经由 Broker Load 写入到 Doris 集群中，并运用 Doris 进行查询加速，以提高咱们对外供给服务的功率。
挑选 Doris 的原因
依据 Apache Doris 高功能、极简易用、实时一致等许多特性，助力 360商业化成功构建了新一代实时数仓架构，本次晋级不仅提高了实时数据的复用性、完成了 OLAP 引擎的一致，而且满意了各大业务场景严苛的数据查询剖析需求，使得全体实时数据流程架构变得简略，大大下降了其保护和运用的本钱。咱们挑选 Doris 作为一致 OLAP 引擎的重要原因大致可归结为以下几点：
- 物化视图： Doris 的物化视图与广告业务场景的特色契合度十分高，比如广告业务中大部分报表的查询维度相对比较固定，运用物化视图的特功可以提高查询的功率，一起 Doris 可以确保物化视图和底层数据的一致性，该特性可协助咱们下降保护本钱的投入。
- 数据一致性： Doris 供给了 Stream Load Label 机制，咱们可经过业务的方法与 Flink 二阶段提交进行结合，以确保幂等写入数据，别的咱们经过自研 Flink Sink Doris 组件，完成了数据的端到端的一致性，确保了数据的精确性。
- SQL 协议兼容：Doris 兼容 MySQL 协议，支持规范 SQL，这无论是关于开发同学，还是数据剖析、产品同学，都可以完成无本钱衔接，相关同学直接运用 SQL 就可以进行查询，运用门槛很低，为公司节省了很多训练和运用本钱，一起也提高了工作功率。
- 优异的查询功能： Apache Doris 已全面完成向量化查询引擎，使 Doris 的 OLAP 功能表现更加强悍，在多种查询场景下都有十分显着的功能提高，可极大优化了报表的询速度。一起依托列式存储引擎、现代的 MPP 架构、预聚合物化视图、数据索引的完成，在低推迟和高吞吐查询上，都到达了极速功能
- 运维难度低： Doris 关于集群和和数据副本管理上做了很多自动化工作，这些投入使得集群运维起来十分的简略，近乎于完成零门槛运维。
在 AB 试验渠道的详细落地
Apache Doris 现在广泛运用于 360商业化内部的多个业务场景。比如在实时大盘场景中，咱们运用 Doris 的 Aggregate 模型对恳求、曝光、点击、转化等多个实时流进行事实表的 Join ；依托 Doris 业务特性确保数据的一致性；经过多个物化视图，提早依据报表维度聚合数据、提高查询速度，由于物化视图和 Base 表的一致关系由 Doris 来保护确保，这也极大的下降了运用复杂度。比如在账户实时消费场景中，咱们首要凭借 Doris 优异的查询优化器，经过 Join 来计算同环比……
接下来仅以 AB 试验渠道这一典型业务场景为例，翔实的为咱们介绍 Doris 在该场景下的落地实践，在上述所举场景中的运用将不再赘述。
AB 试验在广告场景中的运用十分广泛，是衡量规划、算法、模型、战略对产品目标提高的重要东西，也是精细化运营的重要手段，咱们可以经过 AB试验渠道对迭代方案进行测验，并结合数据进行剖析和验证，从而优化产品方案、提高广告作用。
在文章开端也有简略介绍，AB 试验场景所承载的业务相对比较复杂，这儿再详细阐明一下：
- 各维度之间组合灵敏度很高，例如需要对从 DSP 到流量类型再到广告位置等十几个维度进行剖析，完成从恳求、竞价、曝光、点击、转化等几十个目标的完整流量漏斗。
- 数据量巨大，日均流量可以到达百亿等级，峰值可达百万OPS（Operations Per Second），一条流量或许包括几十个试验标签 ID。
依据以上特色，咱们在 AB试验场景中一方面需要确保数据算的快、数据推迟低、用户查询数据快，另一方面也要确保数据的精确性，确保数据不丢不重。
数据落地
当面临一条流量或许包括几十个试验标签 ID 的状况时，从剖析角度动身，只需要选中一个试验标签和一个对照试验标签进行剖析；而假如经过like的方法在几十个试验标签中去匹配选中的试验标签，完成功率就会十分低。
开端咱们希望从数据入口处将试验标签打散，将一条包括 20 个试验标签的流量拆分为 20 条只包括一个试验标签的流量，再导入 Doris 的聚合模型中进行数据剖析。而在这个进程中咱们遇到一个显着的问题，当数据被打散之后会胀大数十倍，百亿级数据将胀大为千亿级数据，即便 Doris 聚合模型会对数据再次紧缩，但这个进程会对集群形成极大的压力。因而咱们抛弃该完成方法，开端测验将压力分摊一部分到计算引擎，这儿需要注意的是，假如将数据直接在 Flink 中打散，当 Job 大局 Hash 的窗口来 Merge 数据时，胀大数十倍的数据也会带来几十倍的网络和 CPU 耗费。
接着咱们开端第三次测验，这次测验咱们考虑在 Flink 端将数据拆分后立刻进行 Local Merge，在同一个算子的内存中开一个窗口，先将拆分的数据进行一层聚合，再经过 Job 大局 Hash 窗口进行第二层聚合，因为 Chain 在一起的两个算子在同一个线程内，因而可以大幅下降胀大后数据在不同算子之间传输的网络耗费。该方法经过两层窗口的聚合，再结合 Doris 的聚合模型，有用下降了数据的胀大程度，其次咱们也同步推进实业务方定时清理已下线的试验，减少计算资源的糟蹋。
考虑到 AB试验剖析场景的特色，咱们将试验 ID 作为 Doris 的第一个排序字段，运用前缀索引可以很快定位到目标查询的数据。别的依据常用的维度组合建立物化视图，进一步缩小查询的数据量，Doris 物化视图根本可以掩盖 80% 的查询场景，咱们会定时剖析查询 SQL 来调整物化视图。终究咱们经过模型的规划、前缀索引的运用，结合物化视图才干，使大部分试验查询结果可以完成秒级返回。
数据一致性确保
数据的精确性是 AB试验渠道的基础，当算法团队呕心沥血优化的模型使广告作用提高了几个百分点，却因数据丢掉看不出试验作用，这样的结果确实无法令人承受，一起这也是咱们内部不允许出现的问题。那么咱们该如何防止数据丢掉、确保数据的一致性呢？
自研 Flink Sink Doris 组件
咱们内部已有一套 Flink Stream API 脚手架，因而凭借 Doris 的幂等写特性和 Flink 的二阶段提交特性，自研了 Sink To Doris 组件，确保了数据端到端的一致性，并在此基础上新增了异常状况的数据确保机制。
在 Doris 0.14 版别中（初期运用的版别），咱们一般经过“同一个 Label ID 只会被写入一次”的机制来确保数据的一致性；在 Doris 1.0 版别之后，经过 “Doris 的业务结合 Flink 二阶段提交”的机制来确保数据的一致性。这儿详细共享运用 Doris 1.0 版别之后，经过 “Doris 的业务结合 Flink 二阶段提交”机制确保数据的一致性的原理与完成。
在 Flink 中做到数据端到端的一致性有两种方法，一种为经过至少一次结合幂等写，一种为经过刚好一次的二阶段业务。
如右图所示，咱们首先在数据写入阶段先将数据写入本地文件，一阶段进程中将数据预提交到 Doris，并保存业务 ID 到状态，假如 Checkpoint 失利，则手动抛弃 Doris 业务；假如 Checkpoint 成功，则在二阶段进行业务提交。关于二阶段提交重试屡次仍然失利的数据，将供给数据以及业务 ID 保存到 HDFS 的选项，经过 Broker Load 进行手动康复。为了防止单次提交数据量过大，而导致 Stream Load 时长超越 Flink Checkpoint 时刻的状况，咱们供给了将单次 Checkpoint 拆分为多个业务的选项。终究成功经过二阶段提交的机制完成了对数据一致性的确保。
运用展现
下图为 Sink To Doris 的详细运用，全体东西屏蔽了 API 调用以及拓扑流的组装，只需要经过简略的装备即可完成 Stream Load 到 Doris 的数据写入 。
集群监控
在集群监控层面，咱们采用了社区供给的监控模板，从集群目标监控、主机目标监控、数据处理监控三个方面动身来搭建 Doris 监控体系。其中集群目标监控和主机目标监控首要依据社区监控阐明文档进行监控，以便咱们查看集群全体运行的状况。除社区供给的模板之外，咱们还新增了有关 Stream Load 的监控目标，比如对当时 Stream Load 数量以及写入数据量的监控，如下图所示：
除此之外，咱们对数据写入 Doris 的时长以及写入的速度也比较关注，依据本身业务的需求，咱们对使命写入数据速度、处理数据耗时等数据处理相关目标进行监控，协助咱们及时发现数据写入和读取的异常状况，凭借公司内部的报警渠道进行监控告警，报警方法支持电话、短信、推推、邮件等
总结与规划
现在 Apache Doris 首要运用于广告业务场景，已有数十台集群机器，掩盖近 70% 的实时数据剖析场景，完成了全量离线试验渠道以及部分离线 DWS 层数据查询加速。当时日均新增数据规划可以到达百亿等级，在大部分实时场景中，其查询推迟在 1s 内。一起，Apache Doris 的成功落地使得咱们完成了实时数仓在 OLAP 引擎上的一致。Doris 优异的剖析功能及简略易用的特色，也使得数仓架构更加简洁。
未来咱们将对 Doris 集群进行扩展，依据业务优先级进行资源阻隔，完善资源管理机制，并方案将 Doris 运用到 360商业化内部更广泛的业务场景中，充分发挥 Doris 在 OLAP 场景的优势。终究咱们将更加深化的参加到 Doris 社区中来，积极回馈社区，与 Doris 并肩同行，共同进步！
