使用统一数据仓库打破数据孤岛：基于Apache Doris的CDP

使用统一数据仓库打破数据孤岛：基于Apache Doris的CDP
随着企业数据来源日益多样化，数据孤岛问题变得普遍。保险公司在构建客户数据平台（CDP）时，面临着数据孤岛导致的组件密集型计算层，数据存储分散的问题。为了解决这些问题，他们采用了基于 Apache Doris 的 CDP 2.0，利用 Doris 的统一数据仓库能力，打破数据孤岛，简化数据处理管道，提升数据处理效率。
随着企业数据来源日益多样化，数据孤岛问题变得普遍。保险公司在构建客户数据平台（cdp）时，面临着数据孤岛导致的组件密集型计算层，数据存储分散的问题。为了解决这些问题，他们采用了基于 apache doris 的 cdp 2.0，利用 doris 的统一数据仓库能力，打破数据孤岛，简化数据处理管道，提升数据处理效率。

数据孤岛问题就像在线企业的关节炎，因为几乎每个人随着年龄的增长都会遇到这个问题。企业通过网站、移动应用、H5 页面和终端设备与客户互动。出于某种原因，整合所有这些来源的数据是一件很棘手的事情。数据保留在原处，无法相互关联以进行进一步分析。这就是数据孤岛的形成方式。您的业务规模越大，您拥有的客户数据源就越多样化，您就越有可能陷入数据孤岛。 

这正是我在这篇文章中要讨论的保险公司所发生的情况。到2024年，他们已服务超过5亿客户，签订了570亿份保险合同。当他们开始构建客户数据平台（CDP）来适应如此大的数据规模时，他们使用了多个组件。 

CDP 中的数据孤岛

与大多数数据平台一样，他们的 CDP 1.0 具有批处理管道和实时流管道。离线数据通过 Spark 作业加载到 Impala，在那里被标记并分成组。同时，Spark 还将其发送到 NebulaGraph 进行 OneID 计算（本文稍后详细阐述）。另一方面，实时数据被Flink打上标签，然后存储在HBase中，以备查询。

这导致了 CDP 中的组件密集型计算层：Impala、Spark、NebulaGraph 和 HBase。

结果，离线标签、实时标签和图形数据分散在多个组件中。由于冗余存储和大量数据传输，将它们集成以提供进一步的数据服务成本高昂。更重要的是，由于存储的差异，他们不得不扩大CDH集群和NebulaGraph集群的规模，增加了资源和维护成本。

基于 Apache Doris 的 CDP

对于CDP 2.0，他们决定引入一个统一的解决方案来收拾残局。在CDP 2.0的计算层，Apache Doris承担实时和离线数据存储和计算。 

为了摄取离线数据，他们利用流加载方法。他们的 30 线程摄取测试表明，它每秒可以执行超过 300,000 次更新插入。为了加载实时数据，他们结合使用了Flink-Doris-Connector和 Stream Load。此外，在需要从多个外部数据源提取数据的实时报告中，他们利用多目录功能进行联合查询。 

此 CDP 上的客户分析工作流程如下。首先，他们整理客户信息，然后为每个客户贴上标签。他们根据标签对客户进行分组，进行更有针对性的分析和操作。

接下来，我将深入研究这些工作负载并向您展示 Apache Doris 如何加速它们。 

一个ID

当您的产品和服务有不同的用户注册系统时，您是否遇到过这种情况？您可以从一个产品网页收集用户 ID A 的电子邮件，然后从另一个产品网页收集用户 ID B 的社会安全号码。然后您会发现 UserID A 和 UserID B 实际上属于同一个人，因为他们使用相同的电话号码。

这就是 OneID 作为一个想法出现的原因。就是将所有业务线的用户注册信息汇集到Apache Doris中一张大表中，进行整理，并确保每个用户拥有唯一的OneID。 

这就是他们如何利用 Apache Doris 中的功能来确定哪些注册信息属于同一用户。

标签服务

该CDP容纳了5亿客户的信息，这些信息来自500多个源表，总共附加了2000多个标签。

按照时效性，标签可以分为实时标签和离线标签。实时标签由 Apache Flink 计算并写入 Apache Doris 中的平面表，而离线标签由 Apache Doris 计算，因为它们源自 Doris 中的用户属性表、业务表和用户行为表。以下是该公司在数据标记方面的最佳实践： 

1. 离线标签

在数据写入高峰期，由于数据规模庞大，全量更新很容易导致OOM错误。为了避免这种情况，他们利用Apache Doris 的INSERT INTO SELECT功能并启用部分列更新。这将大大减少内存消耗并在数据加载过程中保持系统稳定性。

set enable_unique_key_partial_update=true;insert into tb_label_result(one_id, labelxx) select one_id, label_value as labelxxfrom .....