公众号制作模板网站,钢材料 网站建设 中企动力,使用wordpress版权,设计制作效果图简介#xff1a; 友信金服公司推行全域的数据体系战略#xff0c;通过打通和整合集团各个业务线数据#xff0c;利用大数据、人工智能等技术构建统一的数据资产#xff0c;如 ID-Mapping、用户标签等。友信金服用户画像项目正是以此为背景成立#xff0c;旨在实现“数据驱…简介 友信金服公司推行全域的数据体系战略通过打通和整合集团各个业务线数据利用大数据、人工智能等技术构建统一的数据资产如 ID-Mapping、用户标签等。友信金服用户画像项目正是以此为背景成立旨在实现“数据驱动业务与运营”的集团战略。目前该系统支持日处理数据量超 10 亿接入上百种合规数据源。作者 | 杨毅穆超峰贺小兵胡夕导读当今生活节奏日益加快企业面对不断增加的海量信息其信息筛选和处理效率低下的困扰与日俱增。由于用户营销不够细化企业 App 中许多不合时宜或不合偏好的消息推送很大程度上影响了用户体验甚至引发了用户流失。在此背景下友信金服公司推行全域的数据体系战略通过打通和整合集团各个业务线数据利用大数据、人工智能等技术构建统一的数据资产如 ID-Mapping、用户标签等。友信金服用户画像项目正是以此为背景成立旨在实现“数据驱动业务与运营”的集团战略。目前该系统支持日处理数据量超 10 亿接入上百种合规数据源。一、技术选型传统基于 Hadoop 生态的离线数据存储计算方案已在业界大规模应用但受制于离线计算的高时延性越来越多的数据应用场景已从离线转为实时。这里引用一张表格对目前主流的实时计算框架做个对比。Apache Storm 的容错机制需要对每条数据进行应答ACK因此其吞吐量备受影响在数据大吞吐量的场景下会有问题因此不适用此项目的需求。Apache Spark 总体生态更为完善且在机器学习的集成和应用性暂时领先但 Spark 底层还是采用微批Micro Batching处理的形式。Apache Flink 在流式计算上有明显优势首先其流式计算属于真正意义上的单条处理即每一条数据都会触发计算。在这一点上明显与 Spark 的微批流式处理方式不同。其次Flink 的容错机制较为轻量对吞吐量影响较小使得 Flink 可以达到很高的吞吐量。最后 Flink 还拥有易用性高部署简单等优势。相比之下我们最终决定采用基于 Flink 的架构方案。二、用户画像业务架构用户画像系统目前为集团线上业务提供用户实时标签数据服务。为此我们的服务需要打通多种数据源对海量的数字信息进行实时不间断的数据清洗、聚类、分析从而将它们抽象成标签并最终为应用方提供高质量的标签服务。在此背景下我们设计用户画像系统的整体架构如下图所示整体架构分为五层接入层接入原始数据并对其进行处理如 Kafka、Hive、文件等。计算层选用 Flink 作为实时计算框架对实时数据进行清洗关联等操作。存储层对清洗完成的数据进行数据存储我们对此进行了实时用户画像的模型分层与构建将不同应用场景的数据分别存储在如 PhoenixHBaseHDFSKafka 等。服务层对外提供统一的数据查询服务支持从底层明细数据到聚合层数据的多维计算服务。应用层以统一查询服务对各个业务线数据场景进行支撑。目前业务主要包含用户兴趣分、用户质量分、用户的事实信息等数据。三、用户画像数据处理流程在整体架构设计方案设计完成之后我们针对数据也设计了详尽的处理方案。在数据处理阶段鉴于 Kafka 高吞吐量、高稳定性的特点我们的用户画像系统统一采用 Kafka 作为分布式发布订阅消息系统。数据清洗阶段利用 Flink 来实现用户唯一性识别、行为数据的清洗等去除冗余数据。这一过程支持交互计算和多种复杂算法并支持数据实时 / 离线计算。目前我们数据处理流程迭代了两版具体方案如下1.0 版数据处理流程数据接入、计算、存储三层处理流程整体数据来源包含两种历史数据从外部数据源接入的海量历史业务数据。接入后经过 ETL 处理进入用户画像底层数据表。实时数据从外部数据源接入的实时业务数据如用户行为埋点数据风控数据等。根据不同业务的指标需求我们直接从集团数据仓库抽取数据并落入 Kafka或者直接从业务端以 CDCCapture Data Change的方式写入 Kafka。在计算层数据被导入到 Flink 中通过 DataStream 生成 ID-Mapping、用户标签碎片等数据然后将生成数据存入 JanusGraphJanusGraph 是以 HBase 作为后端存储的图数据库介质与 Kafka并由 Flink 消费落入 Kafka 的用户标签碎片数据进行聚合生成最新的用户标签碎片用户标签碎片是由用户画像系统获取来自多种渠道的碎片化数据块处理后生成的。数据服务层处理流程服务层将存储层存储的用户标签碎片数据通过 JanusGraph Spark On Yarn 模式执行 TinkerPop OLAP 计算生成全量用户 Yids 列表文件。Yid 是用户画像系统中定义的集团级用户 ID 标识。结合 Yids 列表文件在 Flink 中批量读取 HBase 聚合成完整用户画像数据生成 HDFS 文件再通过 Flink 批量操作新生成的数据生成用户评分预测标签将用户评分预测标签落入 Phoenix之后数据便可通过统一数据服务接口进行获取。下图完整地展示了这一流程。ID-Mapping 数据结构为了实现用户标签的整合用户 ID 之间的强打通我们将用户 ID 标识看成图的顶点、ID pair 关系看作图的边比如已经识别浏览器 Cookie 的用户使用手机号登陆了公司网站就形成了对应关系。这样所有用户 ID 标识就构成了一张大图其中每个小的连通子图 / 连通分支就是一个用户的全部标识 ID 信息。ID-Mapping 数据由图结构模型构建图节点包含 UserKey、Device、IdCard、Phone 等类型分别表示用户的业务 ID、设备 ID、身份证以及电话等信息。节点之间边的生成规则是通过解析数据流中包含的节点信息以一定的优先级顺序进行节点之间的连接从而生成节点之间的边。比如识别了用户手机系统的 Android_ID之后用户使用邮箱登陆了公司 App在系统中找到了业务线 UID 就形成了和关系的 ID pair然后系统根据节点类型进行优先级排序生成 Android_ID、mail、UID 的关系图。数据图结构模型如下图所示Gephi1.0 版本数据处理流程性能瓶颈1.0 版本数据处理流程在系统初期较好地满足了我们的日常需求但随着数据量的增长该方案遇到了一些性能瓶颈首先这版的数据处理使用了自研的 Java 程序来实现。随着数据量上涨自研 JAVA 程序由于数据量暴增导致 JVM 内存大小不可控同时它的维护成本很高因此我们决定在新版本中将处理逻辑全部迁移至 Flink 中。其次在生成用户标签过程中ID-Mapping 出现很多大的连通子图如下图所示。这通常是因为用户的行为数据比较随机离散导致部分节点间连接混乱。这不仅增加了数据的维护难度也导致部分数据被“污染”。另外这类异常大的子图会严重降低 JanusGraph 与 HBase 的查询性能。Gephi最后该版方案中数据经 Protocol BufferPB序列化之后存入 HBase这会导致合并 / 更新用户画像标签碎片的次数过多使得一个标签需要读取多次 JanusGraph 与 HBase这无疑会加重 HBase 读取压力。此外由于数据经过了 PB 序列化使得其原始存储格式不可读增加了排查问题的难度。鉴于这些问题我们提出了 2.0 版本的解决方案。在 2.0 版本中我们通过利用 HBase 列式存储、修改图数据结构等优化方案尝试解决以上三个问题。2.0 版数据处理流程版本流程优化点如下图所示2.0 版本数据处理流程大部分承袭了 1.0 版本。新版本数据处理流程在以下几个方面做了优化2.0 版本数据处理流程历史数据的离线补录方式由 JAVA 服务变更为使用 Flink 实现。优化用户画像图数据结构模型主要是对边的连接方式进行了修改。之前我们会判断节点的类型并根据预设的优先级顺序将多个节点进行连接新方案则采用以 UserKey 为中心的连接方式。做此修改后之前的大的连通子图图 6优化为下面的小的连通子图图 8同时解决了数据污染问题保证了数据准确性。另外1.0 版本中一条数据需要平均读取十多次 HBase 的情况也得到极大缓解。采用新方案之后平均一条数据只需读取三次 HBase从而降低 HBase 六七倍的读取压力此处优化是数据计算层优化。Gephi旧版本是用 Protocol Buffer 作为用户画像数据的存储对象生成用户画像数据后作为一个列整体存入 HBase。新版本使用 Map 存储用户画像标签数据Map 的每对 KV 都是单独的标签KV 在存入 HBase 后也是单独的列。新版本存储模式利用 HBase 做列的扩展与合并直接生成完整用户画像数据去掉 Flink 合并 / 更新用户画像标签过程优化数据加工流程。使用此方案后存入 HBase 的标签数据具备了即席查询功能。数据具备即席查询是指在 HBase 中可用特定条件直接查看指定标签数据详情的功能它是数据治理可以实现校验数据质量、数据生命周期、数据安全等功能的基础条件。在数据服务层我们利用 Flink 批量读取 HBase 的 Hive 外部表生成用户质量分等数据之后将其存入 Phoenix。相比于旧方案中 Spark 全量读 HBase 导致其读压力过大从而会出现集群节点宕机的问题新方案能够有效地降低 HBase 的读取压力。经过我们线上验证新方案对 HBase 的读负载下降了数十倍此处优化与 2 优化不同属于服务层优化。四、问题目前线上部署的用户画像系统中的数据绝大部分是来自于 Kafka 的实时数据。随着数据量越来越多系统的压力也越来越大以至于出现了 Flink 背压与 Checkpoint 超时等问题导致 Flink 提交 Kafka 位移失败从而影响了数据一致性。这些线上出现的问题让我们开始关注 Flink 的可靠性、稳定性以及性能。针对这些问题我们进行了详细的分析并结合自身的业务特点探索并实践出了一些相应的解决方案。CheckPointing 流程分析与性能优化方案CheckPointing 流程分析下图展示了 Flink 中 checkpointing 执行流程图Flink 中 checkpointing 执行流程Coordinator 向所有 Source 节点发出 Barrier。Task 从输入中收到所有 Barrier 后将自己的状态写入持久化存储中并向自己的下游继续传递 Barrier。当 Task 完成状态持久化之后将存储后的状态地址通知到 Coordinator。当 Coordinator 汇总所有 Task 的状态并将这些数据的存放路径写入持久化存储中完成 CheckPointing。性能优化方案通过以上流程分析我们通过三种方式来提高 Checkpointing 性能。这些方案分别是选择合适的 Checkpoint 存储方式合理增加算子Task并行度缩短算子链Operator Chains长度选择合适的 Checkpoint 存储方式CheckPoint 存储方式有 MemoryStateBackend、FsStateBackend 和 RocksDBStateBackend。由官方文档可知不同 StateBackend 之间的性能以及安全性是有很大差异的。通常情况下MemoryStateBackend 适合应用于测试环境线上环境则最好选择 RocksDBStateBackend。这有两个原因首先RocksDBStateBackend 是外部存储其他两种 Checkpoint 存储方式都是 JVM 堆存储。受限于 JVM 堆内存的大小Checkpoint 状态大小以及安全性可能会受到一定的制约其次RocksDBStateBackend 支持增量检查点。增量检查点机制Incremental Checkpoints仅仅记录对先前完成的检查点的更改而不是生成完整的状态。与完整检查点相比增量检查点可以显著缩短 checkpointing 时间但代价是需要更长的恢复时间。合理增加算子Task并行度Checkpointing 需要对每个 Task 进行数据状态采集。单个 Task 状态数据越多则 Checkpointing 越慢。所以我们可以通过增加 Task 并行度减少单个 Task 状态数据的数量来达到缩短 CheckPointing 时间的效果。缩短算子链Operator Chains长度Flink 算子链Operator Chains越长Task 也会越多相应的状态数据也就更多Checkpointing 也会越慢。通过缩短算子链长度可以减少 Task 数量从而减少系统中的状态数据总量间接的达到优化 Checkpointing 的目的。下面展示了 Flink 算子链的合并规则上下游的并行度一致下游节点的入度为 1上下游节点都在同一个 Slot Group 中下游节点的 Chain 策略为 ALWAYS上游节点的 Chain 策略为 ALWAYS 或 HEAD两个节点间数据分区方式是 Forward用户没有禁用 Chain基于以上这些规则我们在代码层面上合并了相关度较大的一些 Task使得平均的操作算子链长度至少缩短了 60%~70%。Flink 背压产生过程分析及解决方案背压产生过程分析在 Flink 运行过程中每一个操作算子都会消费一个中间 / 过渡状态的流并对它们进行转换然后生产一个新的流。这种机制可以类比为Flink 使用阻塞队列作为有界的缓冲区。跟 Java 里阻塞队列一样一旦队列达到容量上限处理速度较慢的消费者会阻塞生产者向队列发送新的消息或事件。下图展示了 Flink 中两个操作算子之间的数据传输以及如何感知到背压的首先Source 中的事件进入 Flink 并被操作算子 1 处理且被序列化到 Buffer 中然后操作算子 2 从这个 Buffer 中读出该事件。当操作算子 2 处理能力不足的时候操作算子 1 中的数据便无法放入 Buffer从而形成背压。背压出现的原因可能有以下两点下游算子处理能力不足数据发生了倾斜。背压解决方案实践中我们通过以下方式解决背压问题。首先缩短算子链会合理的合并算子节省出资源。其次缩短算子链也会减少 Task线程之间的切换、消息的序列化 / 反序列化以及数据在缓冲区的交换次数进而提高系统的整体吞吐量。最后根据数据特性将不需要或者暂不需要的数据进行过滤然后根据业务需求将数据分别处理比如有些数据源需要实时的处理有些数据是可以延迟的最后通过使用 keyBy 关键字控制 Flink 时间窗口大小在上游算子处理逻辑中尽量合并更多数据来达到降低下游算子的处理压力。优化结果经过以上优化在每天亿级数据量下用户画像可以做到实时信息实时处理并无持续背压Checkpointing 平均时长稳定在 1 秒以内。五、未来工作的思考和展望端到端的实时流处理目前用户画像部分数据都是从 Hive 数据仓库拿到的数据仓库本身是 T1 模式数据延时性较大所以为了提高数据实时性端到端的实时流处理很有必要。端到端是指一端采集原始数据另一端以报表 / 标签 / 接口的方式对这些对数进行呈现与应用连接两端的是中间实时流。在后续的工作中我们计划将现有的非实时数据源全部切换到实时数据源统一经过 Kafka 和 Flink 处理后再导入到 Phoenix/JanusGraph/HBase。强制所有数据源数据进入 Kafka 的一个好处在于它能够提高整体流程的稳定性与可用性首先 Kafka 作为下游系统的缓冲可以避免下游系统的异常影响实时流的计算起到“削峰填谷”的作用其次Flink 自 1.4 版本开始正式支持与 Kafka 的端到端精确一次处理语义在一致性方面上更有保证。作者介绍杨毅友信金服计算平台部 JAVA 工程师穆超峰友信金服计算平台部数据开发高级工程师贺小兵友信金服计算平台部数据开发工程师胡夕友信金服计算平台部技术总监存储 消息中间件 分布式计算 搜索推荐 Java Kafka 分布式数据库 数据处理 流计算 Hbase版权声明本文中所有内容均属于阿里云开发者社区所有任何媒体、网站或个人未经阿里云开发者社区协议授权不得转载、链接、转贴或以其他方式复制发布/发表。申请授权请邮件developerteamlist.alibaba-inc.com已获得阿里云开发者社区协议授权的媒体、网站在转载使用时必须注明稿件来源阿里云开发者社区原文作者姓名违者本社区将依法追究责任。 如果您发现本社区中有涉嫌抄袭的内容欢迎发送邮件至developer2020service.aliyun.com 进行举报并提供相关证据一经查实本社区将立刻删除涉嫌侵权内容。
