1、用户行为分析大用户行为分析大数据数据平台平台演进与经验分享演进与经验分享分享要点 平台发展历程2015-初步尝试2016-快速成长 2017-初步成型01020304分享要点 平台发展历程2015-初步尝试2016-快速成长 2017-初步成型01020304平台发展历程2015集群搭建CDH集群与工信部合作工信部行为分析2016CDH-HDP数据规范制定与造价BG合作接入造价数据建立相关数据仓库接入CRM、授权等2017集群扩容规范推广权限管理深化数据应用体系国际化数据接入施工数据接入分享要点 平台发展历程2015-初步尝试2016-快速成长 2017-初步成型01020304CDH集群集群
2、2015-初步尝试ETL程序(python)日志接受服务工部部会员中心日志数据ETL清洗HIVEHIVE清清洗洗过过滤滤合合并并输输出出数据数据集市集市数数据据服服务务ETL程序(python)业务数据流行为数据流作业调度JOINJOIN数据仓库HBASEHBASEYARNYARN2015-架构特点平台架构特点：平台架构特点：计算逻辑全计算逻辑全sql化，简单易于维护化，简单易于维护平台架构缺点：平台架构缺点：资源利用不充分资源利用不充分 计算任务耗时长计算任务耗时长 无实时处理能力无实时处理能力 单点问题单点问题 监控问题监控问题 平台优化，稳定性问题平台优化，稳定性问题CDH集群2015-
3、初步尝试ETL程序(python)日志接受服务工部部会员中心日志数据ETL清洗HIVE清清洗洗过过滤滤合合并并输输出出数据集市数据服务ETL程序(python)作业调度JOIN数据仓库HbaseYARN12CASE-01 行为数据分析算法背景：没有建立主数据仓库，导致主数据的处理和行为数据交叉处理，计算 量大，逻辑复杂，难以排错。处理流程：主数据入HIVE仓库主数据处理 行为数据入HIVE仓库行为数据处理 HIVE仓库挂接主数据外挂主数据 分析计算业务指标计算CASE-01 行为数据分析算法面面临临问问题题大数据量大数据量join，大量数据跨节点交换大量数据跨节点交换计算时间超长计算时间超长主
4、主数据和行为数据耦合数据和行为数据耦合度度太高太高Hive数据变更困难数据变更困难CASE-01 行为数据分析算法优化方案：建立主数据仓库 单独处理主数据处理 行为数据入HIVE仓库行为数据处理 按业务指标聚合行为数据 分析计算挂接主数据优点优点先聚合后关联，数据量大幅减低先聚合后关联，数据量大幅减低减少减少需要需要shuffle的数据的数据行为数据能获取最新的行为数据能获取最新的主主数据数据CASE-02 Hbase数据载入 Hive sql 计算结果存储Hive外部表跑批处理 Python 程序读取hive外部表数据读取计算结果 Python 调用 api 逐批写入 10000条每批写入H
5、BASE背景：每天增量行为数据入Hbase，前端分析查询处理流程：CASE-02 Hbase数据载入面面临临问问题题Python 读取读取hive结果，需拉取结果至本地结果，需拉取结果至本地单节点写入，性能差单节点写入，性能差数据量大，逐批写入耗时长数据量大，逐批写入耗时长Hbase压力问题（压力问题（hbase 无预分区）无预分区）CASE-02 Hbase数据载入优化方案：Map01Map02Map03reduce01reduce02HfilebulkloadHbase优点优点MR分布式分布式生成生成Hfile,并行并行,提高效率提高效率bulkload,一次性导入一次性导入,Hbase压
6、力小压力小Bulkload适用场适用场景景大批量数据写入大批量数据写入一次写入，频繁读取一次写入，频繁读取分享要点 平台发展历程2015-初步尝试2016-快速成长 2017-初步成型010203042016-快速成长业务数据业务数据业务数据业务数据业务数据业务数据Informatica主数据仓库主数据仓库客户客户用户用户加密锁加密锁授权授权实名实名数据集市数据集市客户客户用户用户授权授权HDP集群集群HIVE清清洗洗过过滤滤合合并并输输出出HBASETEZCompute EngineSparkSpark StreamingSpark SQLKafka集群集群Kafka01Kafka02Kaf
7、ka03日志数据日志数据日志接受服务日志接受服务HTTPHTTP日志接受服务日志接受服务TCPTCPApolloRabbitMQAzkaban数据服务数据服务IO OLAPIO OLAPIO APIPrestoMonitor2016-快速成长平平台台改改进进建立主数据仓库建立主数据仓库，数据准确性提高，数据准确性提高深度应用深度应用yarn，提高资源利用率，提高资源利用率引入引入Kafka，spark streaming，具有，具有实时数据接入与处理能力实时数据接入与处理能力引入引入azkaban任务调度，调度稳定，可视化，日志可查任务调度，调度稳定，可视化，日志可查引入引入Monitor，实
8、时任务监控，实时任务监控引入引入presto，内部，内部应用应用Hive执行引擎优化，执行引擎优化，Tez替换原生替换原生MR对外提对外提供供轻量轻量OLAP服务和服务和Data API服务服务平台稳定性，平台稳定性，HDFS HA，YARN HA,HBASE HA，Spark HA平台稳定性，平台稳定性，优化优化日志接受服务日志接受服务平台稳定性，平台稳定性，mysql 主从备份，主从备份，redis 主从备份主从备份2016-快速成长平平台台不不足足Kafka集群和集群和HDP集群共享，耦合度高集群共享，耦合度高Hbase 集群和集群和HDP集群共享，耦合度高集群共享，耦合度高ES 集群和
9、集群和HDP集群共享，耦合度高集群共享，耦合度高实时检索日志，受条件限制实时检索日志，受条件限制平台响应外部需求慢平台响应外部需求慢无法对外提供平台即席查询无法对外提供平台即席查询平台缺乏权限及配额管理，无法对外开放平台资源平台缺乏权限及配额管理，无法对外开放平台资源2016-快速成长-成长的烦恼业务数据业务数据业务数据业务数据业务数据业务数据Informatica主数据仓库主数据仓库客户客户用户用户加密锁加密锁授权授权实名实名数据集市数据集市客户客户用户用户授权授权HDP集群集群HIVE清清洗洗过过滤滤合合并并输输出出HBASETEZCompute EngineSparkSpark Stre
10、amingSpark SQLKafka集群集群Kafka01Kafka02Kafka03日志数据日志数据日志接受服务日志接受服务HTTPHTTP日志接受服务日志接受服务TCPTCPApolloRabbitMQAzkaban数据服务数据服务IO OLAPIO OLAPIO APIPrestoMonitor21实时清洗实时清洗3CASE-01 HBASE数据迁移Hbase数据迁移失败处理流程：hbase使用快照使用快照方式对需要迁移方式对需要迁移的表进行备份的表进行备份数据备份数据备份 通过网络将数据通过网络将数据导入到新集群内导入到新集群内数据拷贝数据拷贝 使用快照对表进使用快照对表进行恢复行恢
11、复数据恢复数据恢复 hbase数据迁移后数据迁移后无法查询无法查询问题问题CASE-01 HBASE数据迁移原原因因分分析析数据恢复后手动删除了归档数据导致数据恢复后手动删除了归档数据导致hbase快照只是对元数据进行拷贝并不拷贝快照只是对元数据进行拷贝并不拷贝数据文件数据文件hbase快照在恢复表时是将元数据进行恢复快照在恢复表时是将元数据进行恢复然后将数据文件引用指向归档目录中然后将数据文件引用指向归档目录中CASE-01 HBASE数据迁移经验总结：Hbase备份恢复方式：1.导出/导入2.快照 对比：除了更好地保持一致性外，主要的不同在于导出快照是在HDFS的层面操作的。这意味这Mas
12、ter和域服务器与操作无关。因此不需要为不必要的数据创建缓存空间，也不会有扫描过程因为大量对象创建引起的GC暂停。对于HBase来说主要性能影响就是DataNode额外的网络和磁盘负载。快照方式高效，不影响线上业务。CASE-02 计算任务迁移计算任务迁移导致Ambari管理平台监控异常迁移流程：Python实现 依赖第三方的python包 hbase,thrift 写入结果数据至hbase实时任务实时任务PV HDP管理端python实现 为了统一python环境 将实时任务的依赖的第三方python包装入平台的python2.6环境依赖环境迁移依赖环境迁移 Ambari组件 监控服务dow
13、n 后台组件服务服务Active 管理组件显示服务管理组件显示服务downAmbariCASE-02 计算任务迁移原因分析Ambari监控系统监控系统(python2.6)依赖各依赖各Ambari Agent上报上报RangeServer状态信息，状态信息，Ambari Agent节点本地依赖节点本地依赖python的第三的第三方方包（包（hbase，thrift）获取）获取RangeServer的状态信息。的状态信息。计算任务依赖的第三方包（计算任务依赖的第三方包（hbase，thrift）放入了）放入了ambari依依赖的赖的python环境，导致环境，导致ambari agent 无法无
14、法向向Server报告报告Hbase状态信息状态信息。CASE-02 计算任务迁移解解决决方方案案应用层安装独立的应用层安装独立的pythonpython环境，依赖的第三方包安装到该环境，依赖的第三方包安装到该pythonpython环境下环境下,和和HDPHDP依赖的依赖的python完全隔离完全隔离经经验验教教训训HDP集群环境下集群环境下,我们经常使用我们经常使用python 做数据分析做数据分析,ETL任务等任务等,为了不影响系统和平台的稳定性为了不影响系统和平台的稳定性,建议独立安建议独立安装装python环境环境,和系统自带的和系统自带的python完全隔离完全隔离.2016 平台
15、实时任务 2016年平台架构相对2015年最大的一个改变在于引入了spark streaming 和kafka，使平台具有了实时处理的能力，但是在使用的过程中，我们走了不少的弯路。CASE-03 Streaming走过的弯路在没有完全掌握Streaming 技术的前提下，大规模应用。处理流程：Python实时处理技术研究转化2015年业务为实时任务工时：3个月左右积累业务实时转化Python 线程的方式异步提交所有的Streaming任务测试上线大量实时任务（10几个）优先获得资源任务启动某些任务无法启动集群资源24小时被实时任务占用结果CASE-03 Streaming走过的弯路面面临临问问
16、题题某些实时任务无法启动，获取不了资源？任务扩展？某些实时任务无法启动，获取不了资源？任务扩展？实时任务实时任务24小时运行，集群资源被大量占用，如何跑批处小时运行，集群资源被大量占用，如何跑批处理任务理任务?实时任务输出结果直接写入数据集市（实时任务输出结果直接写入数据集市（mysql），实时写），实时写入入mysql，mysql性能问题？性能问题？为了考虑在有限的资源的情况下，如何保证大量的实时任为了考虑在有限的资源的情况下，如何保证大量的实时任务获取资源公平，并有效运行？自定义实现务获取资源公平，并有效运行？自定义实现kafka到到streaming 的消息转发？的消息转发？CASE-0
17、3 Streaming走过的弯路解决解决方案方案出现的大量问题，迫使我们不得不停下来考虑，出现的大量问题，迫使我们不得不停下来考虑，在没有完全掌握在没有完全掌握Streaming技术的情况下，大规模技术的情况下，大规模应用是否合适？应用是否合适？经经验验教教训训对对Streaming的应用没有经过深入的研究分析的应用没有经过深入的研究分析,使用经使用经验不足验不足Streaming 任务（任务（yarn|standalone）24小时运行，一小时运行，一旦资源持有，不会根据处理的数据量弹性调整资源。旦资源持有，不会根据处理的数据量弹性调整资源。Streaming 相对于跑批来说，资源代价较高，
18、业务没相对于跑批来说，资源代价较高，业务没有实时性要求的话，尽量使用跑批处理。有实时性要求的话，尽量使用跑批处理。Streaming 适用流式数据实时接入适用流式数据实时接入,实时数据清洗，日实时数据清洗，日志数据实时入志数据实时入HBASE|ES等场景等场景CASE-04 Streaming走过的弯路背景：实时计算各产品的用户对功能点的访问情况(PV)处理流程：采用Redis 记录实时消费kafka offset位置 Streaming 初始化从redis获取offset信息获取offset区间段 Spark Streaming初始化kafka参数Spark 初始化Kafka读取参数 采用D
19、ataFrame reducebykey 算子逐批计算转化读取kafka数为DataFrame 利用hbase计数器叠加功能点访问次数结果写入HbaseCASE-04 Streaming走过的弯路面临面临问题问题Kafka和和Spark Streaming 如何集成？保障数据处理高效和稳定。如何集成？保障数据处理高效和稳定。技术技术方案方案1.自定义实现自定义实现kafka到到Streaming之间的消息转发组件之间的消息转发组件(Socket)2.开源组件实现消息转发开源组件实现消息转发,具有根据具有根据Streaming处理数据的时间处理数据的时间调节发送调节发送Kafka数据量大小数据量
20、大小解决解决方案方案1.Spark Streaming 连接连接Kafka 采用采用DirectStream方式方式2.Spark.streaming.kafka.maxRatePerPartition:5000CASE-04 Streaming走过的弯路经验教训：p Spark Streaming 连接kafka有两种方式1.有 receiver 2.无 receiver(DirectStream)对比:a)有receiver方式，revceiver进程启动在分布式节点的一台机械上，Streaming消费的所有数据由该进程转发，效率低下。b)无receiver方式，Streaming 创建的
21、RDD Partition和Kafka Topic Partition 一一映射，即每个RDD Partition 都有一个 Consumer消费对应Kafka Topic Partition数据，性能较高。p 结合Spark.streaming.kafka.maxRatePerPartition参数（每秒钟从kafka Partition 消费的数据），可解决问题。2016 数据清洗2016年在完成集群迁移年在完成集群迁移,架构调整之后架构调整之后,平台已经逐步接入平台已经逐步接入公司多条产品线公司多条产品线(材价产品线材价产品线,造价产品线等造价产品线等)行为数据行为数据,为为了进一步规范
22、数据接入流程了进一步规范数据接入流程,降低数据接入成本降低数据接入成本,我们制定我们制定数据埋点规范数据埋点规范,平台加入了实时校验和转换组件平台加入了实时校验和转换组件,但是面对但是面对多样化的行为数据多样化的行为数据,我们依然遇见各种脏数据我们依然遇见各种脏数据,在接入和转在接入和转换过程中走了不少弯路换过程中走了不少弯路.CASE 05 数据清洗背景:将收集上来的行为数据转换为parquet格式处理流程:每天定时利用python脚本拉取本地文本数据至hive仓库文本数据文本数据入入HIVE仓库仓库 利用Hive自带的Json函数解析文本数据解析解析Json数据数据 利用Hive对parq
23、uet 格式的支持生成parquet 外部表转换转换parquet格式格式 Hive外部表Parquet文件供spark分析结果结果CASE 05 数据清洗面临问题:转换文本数据为parquet格式，部分记录被截断，数据不完整。问题分析:行为数据 keywords、prjfullpath 字段出现“tXXX”字符串解决方案:CASE 05 数据清洗面临问题:在行为数据转换为parquet格式的过程中，外挂主数据的过程中行为数据被放大。该问题目前我只是找到了解决方案，原因还在探索中。问题分析:CASE 05 数据清洗解决方案：CASE 05 数据清洗经经验验教教训训Hive 文本表指定列分隔符和
24、转义符时，要尽可文本表指定列分隔符和转义符时，要尽可能避免指定的分割符和原始记录的字串冲突能避免指定的分割符和原始记录的字串冲突数据进入数据进入hive仓库前，清洗掉仓库前，清洗掉 t n 等特殊字符等特殊字符Hive 表关联条件，避免使用函数对条件字段进表关联条件，避免使用函数对条件字段进行转换，尽可能的使用行转换，尽可能的使用string类型类型CASE 06 Hive Schema无法识别背景:为了减少查询数据量，加快查询速度，对parquet 文件进行分区。面临问题：在hive中使用动态分区字段进行条件查询时，提示分区字段不存在。CASE 06 Hive Schema无法识别分析问题:
25、CASE 06 Hive Schema无法识别解决方案：CASE 06 Hive Schema无法识别经经验验教教训训目前目前hive1.2.0-1.2.1动态分区会识别不到动态分区会识别不到parquet中的分中的分区字段区字段hive对对parquet数据文件索引支持的不是特别好数据文件索引支持的不是特别好关闭关闭hive的索引的索引hive.optimize.index.filter后正常后正常目前目前hive的索引对查询性能的提升有限的索引对查询性能的提升有限分享要点 平台发展历程2015-初步尝试2016-快速成长 2017-初步成型01020304Azkaban业务数据平台架构-2
26、017业务数据业务数据Informatica主数据仓库客户用户加密锁授权实名HDP集群HIVE清洗过滤合并输出HBASETEZSparkSpark StreamingSpark SQLCompute Engine日志数据日志接受服务HTTPKafka集群Kafka01Kafka02Kafka03日志接受服务TCPApolloRabbitMQ数据集市客户用户授权数据服务IO OLAPIO APIMonitorTableauPrestoESranger平台架构-2017平平台台特特点点平台稳定性，基本做到各节点平台稳定性，基本做到各节点HA平台批处理能力平台批处理能力平台实时处理能力平台实时处理能
27、力平台任务调度稳定，可视化平台任务调度稳定，可视化平台全局监控平台全局监控平台化，基本具备对外提供数据计算和存储能力平台化，基本具备对外提供数据计算和存储能力平台即席查询能力平台即席查询能力平台数据多维分析能力平台数据多维分析能力Azkaban业务数据2017-初步成型 业务数据业务数据Informatica主数据仓库客户用户加密锁授权实名HDP集群HIVE清洗过滤合并输出HBASETEZSparkSpark StreamingSpark SQLCompute Engine日志数据日志接受服务HTTPKafka集群Kafka01Kafka02Kafka03日志接受服务TCPApolloRabb
28、itMQ数据集市客户用户授权数据服务IO OLAPIO APIMonitorTableauPrestoESrangerYARN122017-平台优化用户行为大数据平台经过2年的积累和沉淀,已逐渐趋于稳定.同时我们积累了大量的数据和业务,随着业务的增长,资源越来越紧张,我们对平台进行了多方面优化。CASE 01 yarn资源优化优化背景：平台实时PV任务原有资源配置情况如下：CPU核数:1(driver)+5(executor)5(executor-cores)=26核内存:1(driver)1G +5(executor)5G(executor-memory)=26G问题：Yarn 监控列表显示
29、，核数26 正确，内存约41G？CASE 01 yarn资源优化问题分析：yarn 容器最小内存Size：4096M申请 executor-memory 5G 超出4G，因此yarn会给executor 8G的内存。此时每个executor cores 为5，5个线程，8G内存，每个线程堆内存大小1.6G跟踪实时任务日志，我们可以发现executor内存大量闲置。如何优化？优化方案：CASE 01 yarn资源优化经验总结：p 给spark任务分配资源时，应结合yarn容器的最小内存，控制每个executor中线程的内存，根据任务日志，调整executor-cores 和 executor-m
30、emory参数，让每个executor内存最大化利用。p 如果任务涉及RDD persist或cache，应给executor 的存储内存预留一定的空间，如果executor线程过多，executor 执行内存抢占存储内存，占用executor总内存比例过大，persist或cache操作容易造成内存溢出。CASE 02 数据倾斜背景：接口数据产品分区写入hbase，但是产品之间数据量差异巨大处理流程：后台作业每天跑批计算目前我们有12个接口计算所有产品的借口数据后台跑批计算后台跑批计算 调用调用Hbase Api 将结将结果写入果写入Hbase Hbase rowkey 按照按照 产品编号倒
31、排作为前产品编号倒排作为前缀缀计算结果写入计算结果写入HBASE 前端前端WEB API 查询查询Hbase Json 形式返回结果形式返回结果前端前端API查询结果查询结果面临问题：Hbase监控界面发现，部分RangerServer请求热点，极端情况下，RangeServer 服务Down。CASE 02 数据倾斜问问题题分分析析Hbase 表表按照按照0-9预分区预分区Rowkey按照产品编号（数值类型）倒排按照产品编号（数值类型）倒排部分产品数据量太多部分产品数据量太多 CASE 02 数据倾斜平台行为数据分布（2 8 分布）CASE 02 数据倾斜解决方案：我们采用rowkey前打随
32、机数的方式，让大数据量的产品数据均应分布到RangeServer（牺牲前端查询性能）数据抽样，将大数据量产品独立建表CASE 02 数据倾斜经验总结：大数据分析计算时，在数据出现倾斜的情况下，抽样数据，查看数据的分布特征是正太分布还是2 8 分布。2 8 分布的情况下，采用独立处理或者利用其它技术手段（随机数）尽量让数据符合正太分布。适用场景：a)表joinb)聚合计算：reduceByKey,GroupByKey,CombineByKey,FlodByKey等c)Shuffle操作CASE 03 Hbase 二级索引方案当时需求：用户可以根据多条件（gid，dognum，ip，mac等）查询
33、用户 行为数据。处理流程：Spark Streaming 程序实时记录用户行为数据至Hbase Rowkey md5行为数据行为数据入入Hbase Spark Streaming 实时记录 rowkey和gid dognum mac ip等关系入Mysql二级索引二级索引入入Mysql 根据用户传入的条件从mysql二级索引表检索rowkey 根据rowkey获取结果数据前端应用查询前端应用查询CASE 03 Hbase 二级索引方案面临问题：p 索引记录表和行为数据表相同数据量p 当行为数据量越来越大时，Mysql二级索引表性能急剧下降解决方案：p 采用Elasticsearch存储用户的行
34、为数据p Elasticsearch分布式索引机制，支持灵活的多条件查询，无需单独构建二级索引经验总结：p 如果采用hbase二级索引方案，Mysql不合适，大数据量的情况，mysql 查询性能差，可采用华为等第三方二级索引开源技术p 如果 Elasticsearch满足应用场景需求，可采用Elasticsearch存储数据，避免维护二级索引。CASE 04 异构计算方案用户行为大数据平台整个资源分配调度基于用户行为大数据平台整个资源分配调度基于yarn实现实现,yarn能够在能够在cup和和memory上实现较为细粒度的控制上实现较为细粒度的控制(相对于磁盘相对于磁盘IO),IO是整个集群共
35、享的是整个集群共享的。在集群资源使用的高峰期在集群资源使用的高峰期,磁盘磁盘IO将成为瓶颈将成为瓶颈,实时任务这时可能因为磁盘实时任务这时可能因为磁盘IO问问题而失败题而失败.为了保证实时任务的高可用为了保证实时任务的高可用,期望能够用几个独立的物理机来运行集群上的所期望能够用几个独立的物理机来运行集群上的所有实时任务有实时任务,实时任务独享这几台机械的实时任务独享这几台机械的IO,完全物理隔离实时任务完全物理隔离实时任务,达到实时达到实时与跑批的异构计算与跑批的异构计算.CASE 04 异构计算方案实实现现原原理理Yarn Capacity调度器支持基于标签的调度调度器支持基于标签的调度可以将集群节点打上标签可以将集群节点打上标签,将他们按标签分组将他们按标签分组将标签组挂在到指定将标签组挂在到指定Yarn队列队列作业任务提交到指定队列实现分离作业任务提交到指定队列实现分离CASE 04 异构计算方案集群标签分区(分组):集群标签分区(分组):CASE 04 异构计算方案Yarn队列绑定标签:
