蘑菇街MINI DMP系统实践

背景

       大数据时代，DMP这个词越来越多地出现在人们视野内，为了更好地促进数据流通，并利用数据，以技术和数据驱动运营和效果，各大厂都沉淀了自己的DMP系统，比较出名的如阿里的达摩盘，腾讯的广点通DMP等。

        我们的数据很大(5亿总量的记录)，然而不会铺很大的盘，既没有提供复杂炫目的交叉分析，也没有加入高端大气的人群looklike，我们是从一个小小的点出发，定位于利用站内的数据，服务于站内业务的现状，提供适合业务场景的定向投放/推送等活动。目前典型的场景是用于资源位定向，弹窗定向和个性化推送，覆盖到首焦、弹窗、新客流程及其他业务场景的定向业务，为不同的用户定制不同的展示和流程，提升用户的兴趣和体验

系统架构

DMP的核心是由用户画像和DMP服务两部分组成
系统通过高效的数据运转，支撑上层的业务场景

从结构看，我们的DMP如下图所示，由以下几块组成: 用户画像（数据源，数据流，数据存储） DMP业务 (数据同步，任务调度，上层业务)

下面分别从各个维度简单介绍下DMP的技术实现。

用户画像

数据源

DMP的数据源来源于以下几部分：
1. 实时打点日志清洗  -->  生成实时标签 (是否签到，各维度首购等)
2. 统计类型离线数据(BI) --> 离线行为标签
3. 用户宽表(BI)  --> 用户基本信息
4. 策略数据(算法) --> 偏好类标签

这里大部分标签要感谢兄弟团队同学的支持。

数据流

DMP的数据流是一个典型的ETL的过程。
而目前外部的ETL工具或者数据介质扩展不灵活，
或者使用和扩展成本比较高，考虑集成站内的一些组件，
自主开发的灵活性更强，定制难度更低，
因此开发了自己的ETL模块。

ETL是一个异构的数据之间高速交换数据的工具，实现了在多种的数据介质(Hdfs/Hive/Hbase）之间的数据交换。采用disruptor作为引擎，读写端分离，分别使用插件加载的方式来新增读写介质的支持，数据导入导出通过规则配置即可完成。新增1个数据源，我们只需要针对新增的数据源开发的一套Reader/Writer插件，而不用开发多条链路场景，即可实现任意数据的互导。并且数据流节点支持多实例分离和横向扩展。

存储

以Hbase为基础存储，上延到hive，es，
缓存等多种介质的存储体系

HBase

• 选用Hbase作为基础存储主要考虑以下几点：

1. 列可扩展性，并且适合亿级别的数据存储

2. 存储特性，比如ttl机制以cell为单位，未更新数据按照时间消亡。

3. 方便生成外表和准时间查询数据。

用户画像数据在hbase的结构如下图所示: 

ElasticSearch

• 快速获取一定条件下的人群人数

• 可以接受的scroll search获取人群的效率

• 基于标签的聚合分析(人群分析)

• 通过sIREn join插件支持自有人群的聚合分析。

DMP服务

数据同步

数据同步从场景上分为全量，小全量和实时
全量和小全量在实现上比较简单，但是数据量级过大时，
时间和服务器资源消耗较大。 
实时同步的关键在于如何获取变化数据（而不仅是更新数据）

实时同步实现方案

• 如上图所示：

1. 画像系统将特征数据写入Hbase（Master Cluster）；

2. Master集群 通过 replication 机制 异步地将 edit信息同步到slaver集群；

3. slaver集群 通过 Hbase 协处理器 coprocessor 过滤出变化数据，放入中间件队列；

4. DMP consumer 消费掉队列中的任务，转换格式，写入ES

Detail

想法源于mysql的binlog而衍生的同步机制。 对等到hbase，hbase也有Hlog记录了数据的写入动作。 那么如何通过Hbase的Hlog来实现 数据变化的实时感知呢

这里涉及到两个概念：

Hbase replication && coprocessor

• replication

  Hbase 提供了一种集群replication的机制，允许通过WALEdit和hlog来实现把一个集群的状态同步

  到另一个集群。当请求发送给master cluster时，hlog日志放入hdfs的同时进入replication队列，

  由slave cluster通过zookeeper获取并写入slave的表中。一些常用的使用场景可能是备份和故障

  恢复或者数据聚合统计等。这里，我们使用replication机制来实现数据从hbase到ES的实时同步。

  replication的机制如下图： 

（from http://hbase.apache.org/book.html）

replication 有三种模式，主－从， 主－主和循环。 这里的场景比较适合主－从模式，在从集群里生产实际的同步任务，主集群仅仅通过异步的方式传递edit辑信息，异步模式给主集群带来的很小的开销，并且不会对主集群的正常处理产生影响。同步动作以批量进行，可以通过配置一个大小或者时间来设定，或者通过事件来触发，即时同步，根据实际需要来决定。

异步机制如下： 

• （from @猎户）

  以上实现了从master集群同步edit信息到slaver集群，以便使同步的后续动作可以在slaver集群上开展而不用影响主集群的功能。

• coprocessor

  coprocessor使得能够直接在每个region server上执行用户自定义代码，为了更好支持其行为的灵活性，提供了两个不同方面的插件。一个是观察者（observer），类似于关系数据库的触发器。另一个是终端(endpoint)，动态的终端有点像存储过程。

  其中观察者模式下又包括三个模式 RegionObserver：提供各种动作（Get／Put等）的前后置回调钩子 WALObserver：提供WAL相关操作钩子。 MasterObserver：提供DDL-类型的操作钩子。如创建、删除、修改数据表等。

  这里使用 RegionObserver模式，RegionObserver的工作原理如下： 

• （from https://blogs.apache.org/hbase/entry/coprocessor_introduction）

  我们在slaver集群上处理数据变化，具体动作：

• put 前置钩子中比较put值和原值是否一致，以确定是否变化数据。如有变化数据，则生产任务放入kafka/corgi，类型为put

• delete 后置钩子中 生产任务放入kafka/corgi,类型为delete

• Coprocessor并非独立进程，而是在原来的进程中嵌入操作的钩子，其生效也不需要重启系统，仅需要将实现编成的jar包放置在hdfs的coprocessor，通过shell加载。

以上两个机制，可以实现，将master集群的变化数据作为任务放置到kafka/corgi中，后续同步到ES的动作由业务实现，一些适配和第三方接口即可，不仅可以用于同步到ES，还可以用于一些实时的业务处理。

在从底层数据／队列到DMP的这条链路上，如果底层队列本身是实时队列，那么通过这条链路可以实现DMP的实时特征，这一特性对于DMP后续的一些特征和场景还是有很重要意义的，比如最后打开app时间，登录、激活动作／时间等。

• 这里解释下DMP为什么不直接对接实时的数据源

1. 实时数据源的数据属于原始数据，需要清洗整合甚至算法来得出所需要的数据，业务如果直接对接这里意味着大量的重复处理。

2. 业务这边的ES更关注变化的数据以此来更新业务所依赖的ES数据，如果从源头获取，或者hbase数据同级的中间件获取只能获取全量，量级太大。

缺点和现状

• 目前该机制作在量级上受到较大的限制，因为数据刷入的量级过大，对于从集群的要求也会比较高，另外hbase自身的缓存不宜设置过大，导致部分数据丢失。因此对于需要实时订阅变化数据的量级不能太高。

任务调度

DMP系统中需要有大量的长任务来支撑业务,包括采集人群,投放人群等；
对于任务的可靠性和任务节点的可扩展性要求比较高。

这里选择了开源的轻量级任务平台LTS作为任务调度框架。 在实现上，仅需要实现任务提交和执行节点的开发即可，并且任务失败后，可以自动重新分配节点执行。 具体结构和机制如下： 

（from https://github.com/ltsopensource/light-task-scheduler）

业务服务

业务上主要支撑两类场景，定向投放和推送。

以投放为例，业务提供服务的前提是在DMP圈选人群，完成人群圈选和投放后，提供以下服务:

1. . 根据did/userid，获取所属离线人群id

2. . 根据did/userid，获取其特征，匹配已投放的实时人群条件，获取id。

业务方获取人群id，根据投放时配置的策略，展示不同的素材/活动等 

结语

DMP系统是运行在以数据流为基础，基于多态的数据，通过业务需要的调度任务，完成上层对于投放，推送等活动的定向。当然，目前来看，用户的画像模型和数据的关系还比较简单，另外从业界来看，DMP最常用的场景还是外部的DSP投放，并且需要从外界不断补充数据来完善自己的用户画像体系。随着今年外投场景和业务的拓展，这些功能也正在探索和进行中...

更多流量、广告、搜索、算法相关内容, 敬请关注“美丽联合数据技术”公众号