搜公众号
推荐 原创 视频 Java开发 开发工具 Python开发 Kotlin开发 Ruby开发 .NET开发 服务器运维 开放平台 架构师 大数据 云计算 人工智能 开发语言 其它开发 iOS开发 前端开发 JavaScript开发 Android开发 PHP开发 数据库
Lambda在线 > DataFunTalk > 回顾·基于HBase实时数仓探索实践

回顾·基于HBase实时数仓探索实践

DataFunTalk 2019-02-13
举报

回顾·基于HBase实时数仓探索实践

本文根据上海久耶大数据研发工程师武基鹏在中国HBase技术社区第四届MeetUp上海站中分享的《基于HBase实时数仓探索实践》编辑整理而成。

回顾·基于HBase实时数仓探索实践

今天从六个方面介绍,首先是久耶第一代离线数仓以及第二代实时数仓。接下来介绍下公司业务场景和业务开发,基于HBase的开发流程,然后介绍下公司CDH集群,介绍下CDH集群调优监控。最后分享两个生产案例。

回顾·基于HBase实时数仓探索实践

第一代离线数仓是在去年三月份上线,主要是基于OMS和WMS,由于分库分表,大约有十几个库。前期通过SQOOP进行数据抽取,后来由于SQOOP的一些问题采用了阿里开源的DataX,时间粒度使用脚本调度实现T+1模式抽取到Hive。作分析时采用Apache的Kylin,将数据直接存入HBase中,最初数仓建立用于数表查询,用的是Superset,后面也用了Saiku。六个月就被淘汰,因为离线数仓有个问题,业务部分要查询数据,但是你的时间粒度是昨天的,业务部分需要看当天一段时间内的数据,离线是无法满足。

回顾·基于HBase实时数仓探索实践

接下来在另一个集群构建实时数仓,其选型主要有两个,第一个是实时,实时采集利用Maxwell,直接采集公司数据库MySQL,将数据直接以json格式发送到kafka,数仓存储选型是HBase。

为什么选择Maxwell呢,第一个它能够使用“select * from Table”进行bootstrapping初始化数据,在大数据构建时可以利用Maxwell进行全表扫描,这句SQL会自动触发Maxwell某个线程进行数据拉取。第二个Maxwell支持断点还原功能,大数据平台架构不光考虑到高可靠、高性能,也要保证数据零丢失,它支持记录MySQL的post日志进行数据还原,这是当初选择最重要的原因。第三个Maxwell将数据从MySQL发送到Kafka,Kafka是分区的,如何保证全局有序是个问题。它能保证这个特性,支持database, table, primary key, or column的拼接,将数据发送到某个分区;比如一条业务数据在业务系统先在insert再做update再做delete,Kafka会将这三者发送到三个分区,key值为空不会记录,在销毁时用sparkstreaming可能会以delete、update、insert顺序,会造成数据紊乱。我们希望将这些特征数据发送到Kafka一个分区,而Kafka单分区是有序的。第四个Maxwell也会将这些数据发送到后端,当业务数据的表需要升级,如加索引、加字段,可以通过alert语句解析捕获,进行同步更新到HBase中。因此基于这四点要求选择了Maxwell,没有选择当前其他开源产品。

回顾·基于HBase实时数仓探索实践

接下来讲一下为什么选择HBase而不选择pudu等产品。第一个是HBase是分布式、可缩的。第二个是随机的读和写,第三个HBase支持百万列。第三个介绍下为什么要选择Phoenix,首先原因是支持SQL,利用原生HBase进行查询、代码分析比较吃力。第二个我们构建的表是盐表,能够解决热点问题,避免一个节点很繁忙另一个节点很闲。第三Phoenix支持二级索引,由于表是盐表(分区),索引也是分区的。第四个支持Spark,可以直接将表传入Phoenix而不用通过HBase,有利于传统开发人员转型,而不用专注于底层HBase。

回顾·基于HBase实时数仓探索实践

基于CDH HBase版本构建Phoenix版本历程,phoenix-for-cloudera-4.9、HBase-1.2、cdh5.9,这个存在问题,然后采用apache-phoenix-4.11.0、HBase-1.2,最后采用phoenix-for-cloudera-4.10、HBase-1.2、cdh5.12。cdh5.11的邮件配置存在bug。

进行编译的原因是去年Phoenix官方是不支持CDH版本,目前是支持的。编译时将pom文件,改为CDH支持,然后改生产需要的Spark版本。修复SYSTEM.MUTEX表在分布式的计算时,多次创建错误。QueryServicesOptions.java文件修改参数DEFAULT_IS_NAMESPACE_MAPPING_ENABLED=true。Phoenix存在一个问题就是时区,比如一条上午十点的业务数据在Phoenix周转下,时间数据会减一个8小时。修改DateUtil.java文件timezone为”Asia/Shanghai”,但是读写两种只解决了一种,而业务代码开发需要经过Phoenix架构JDBC,数据还是会出错,上面只解决了查询,后来采用下面改动,然后编译。

回顾·基于HBase实时数仓探索实践

上图是实时数仓架构图,主要的存储层还是以HBase为主。第一层业务系统数据库在阿里云平台上,有OMS、WMS,Report DB是OMS和WMS的重复,将里面的数据全部同步于一台机器,使用的就是Maxwell,其支持白名单和黑名单。业务平台的表可能有两三百个,大数据平台的计算可能只需要100多个,可以添加白名单,有些表的数据就可以不用过来了。这些数据通过Json发送到Kafka,然后通过Spark streaming去消费Kafka,通过JDBC写入HBase。表不是通过Phoenix语句创建,不关心底层HBase,只需要通过Phoenix像MYSQL一样查询即可。同时会将计算结果存储到Redis,久耶慧策应用也会将数据写入ES里面。中间一层就是常见应用开发,如Spark Streaming、Spark SQl,也用Python和R语言。调度平台起先用的是Azkaban,然后是Airflow,最后用的是Oozie。上图蓝色是实时大屏,红色是全球仓库,大约有四十几个,数据绑定用的是saiKU,将Phoenix架包集成进去,saiKU分上卷和下卷,业务人员依据自己的需求去拿行和列数据,saiKU通过Phoenix组装SQL语句查询结果数据。也用到zeppelin,这是Spark交互式开发必须用到的。

接下来讲一下数据仓库,首先是模型建设,第一层是基础表,在Phoenix中建立与MySQL一样的表。在基础表的基础上构建事实表(订单实时发生的表)和维度表(如中国有多少省多少市等更新不是很大的表),依据事实表和维度表进行代码开发,构建领域表,就是依据业务需求得出的结果存到领域表。数据校验是通过数据量比对,起先是在重库时做触发器,但是MySQL重库触发器支持不友好。通过改造Phoenix代码将数据写入Redis,增加加一删除减一,MySQL数据和HBase数据是一天一查一对比,当不相等直接调用shell脚本进行全表扫描。当前只采用OMS、WMS的库,QPS处于2000,1条数据: 平均60列 495b。

回顾·基于HBase实时数仓探索实践

业务场景开始是业务报表开发,有客诉妥投、ABC订单、商业季度等。也提供一个BI自助分析,第三个就是双十一大屏和龙虎榜,同时使用了BMS系统,是一个商业结算系统。第五个是今年做的领导层和客户层的慧策,商业决策分析。

业务开发套路就是依据业务需求将数据存在那些表里面,需要将构建表的语句提取出来构建Phoenix Table,然后Kafka+Spark Streaming+Phoenix进行数据的插入。接着就是Spark开发读和写,我们还利用了DBeaver。我们建表使用了联合组件,由于公司集群规模不是很高,regionServer是38台,COMPRESSION 是使用SNAPPY,这是依据压缩比、解压性能。

回顾·基于HBase实时数仓探索实践

接下来是一个经典开发案例Kafka+Spark Streaming+PhoenixPhoenix可以理解为MySQL架包的JDBC。我们并没有使用Phoenix的Pool池,官方也推荐使用正常JDBC文件,因为JDBC已经支持长连接,foreachPartition拿到Phoenix的JDBC,中间进行常见数据处理,Kafka接收过来数据是Json格式,如何将其转化为Phoenix的upset语法和delete语法,完成后就将连接关闭。

回顾·基于HBase实时数仓探索实践

回顾·基于HBase实时数仓探索实践

数据流入Phoenix大数据平台是通过bootstream的全表扫描,其增量数据也是实时进入。业务代码开发首先将架包导入pom文件,如何找维度是将Phoenix的Apache下载到IDEA,在测试类里面查找。Phoenix+Spark读取有好几种,选择以上写法原因有:首先其支持列裁剪,第二支持where条件,configuration指的是Spark的HDFS的conf。

业务开发是多张表,Spark表是df,接下来就和Phoenix和HBase无关。接下来就是对接Spark业务开发逻辑处理,最后结果集会回写到HBase中。还是通过Phoenix写入,有追加、overwrite。HBase没有很好地可视化工具,利用DBeaver,支持MYSQL、Oracle等所有数据库类型,也支持二次开发借助于接口实现。

回顾·基于HBase实时数仓探索实践

接下来介绍下集群调优参数,分为六个方面:(1)Linux parameters、(2)HDFS parameters、(3)HBase parameters、(4)GC parameters、(5)Monitor、(6)Bug。句柄数、文件数、线程数这些都是要调,因为regionserver在操作时需要open file,处理时需要用到一些线程,一些系统都是架设在Linux上,因此集群调优都需要调它。需要注意的是改完后需要检查是否生效,立即生效是sysctt-p。Spark开发需要将数据频繁的写入HBase中,HBase底层是HDFS,在写入时就会出现问题,最后发现Linux系统参数没有调。

回顾·基于HBase实时数仓探索实践

在正常的HBase节点机器上,swap是设置为0,这并不是禁用swap而是其惰性是最大的。由于我们公司由于业务系统较多,吃的内存比较紧,因此设为10,这样可以使job慢一点但是不能挂,但是如果做实时就需要设置为0。这个最终设置取决于你们自身业务环境,选择自己需要的就好。如果做CBH的平台部署必须要关闭大页面。

接下来分享一个有意思的参数HDFS Parameters,正常调优是CBH界面打开、HBase的xml文件打开。主要调优是timeout和handler参数,将其几倍放大,socket.timeout在HBase的xml文件一定要部署,否则无法支持高并发操作。

回顾·基于HBase实时数仓探索实践

当一个本机线程无法创建一个本机线程,这段代码打在HDFS的dataload,当时dataload的内存配置是8G,实际只使用1G,这个时候就休要加上echo "kernel.threads-max=196605"->/etc/sysctl.confecho"kernel.pid_max=196605"->/etc/sysctl.conf,echo "vm.max_map_count=393210"-> /etc/sysctl.conf三个参数,这其实是底层Linux抛出的错误。提醒一点socket.timeout参数不仅在HDFS中需要配置,在HBase中也需要配置。

GC是regionserver配置,但是配置是CDH配置,GC默认垃圾选择器是CMS,需要将其改为GE,如果需要配置可以去尝试下,小米以前分享过。可以对参数进行调试进行压错调优,尤其大数据平台开发尤其如此。

回顾·基于HBase实时数仓探索实践

项目上线需要做监控,第一个就是HBase的读和写,绿色是写,但是读存在两个波峰,因为我们的调度平台以一个小时将所有job调度完。图中Y轴是每秒的请求量,如果写的量上来了或者读的波峰没有规律,就有可能是集群宕了。

第二个监控的指标是FDS,就是Regionserver的文件句柄数,如果请求很多,句柄数会很高,因为其底层依赖于Linux,如果超过Linux设置值机器容易夯住下线,导致CPU不正常,这时需要后台强制机器下线。然后需要监控Zookeeper,监控的是Zookeeper Open Connections,因为HBase进行操作需要打开的连接,当业务场景为长服务,如Spark streaming一直运行,先前尝试用SparkSQL+Phoenix做一个长服务,因为调度都是通过shell脚本调度,在资源紧张时需要抢资源,在submit时需要申请资源(大约30S),线上是不允许的。最后采用Spark streaming+Spark SQL+Phoenix JDBC,Spark streaming是实时的每隔一小时判断进行数据处理,这个时候Zookeeper Open Connections就随着递增趋势上涨,当到Connection数(默认500)CDH会杀掉。后来改为水平,利用PHOENIX-4319:Zookeeper connection should be closed immediately解决问题

回顾·基于HBase实时数仓探索实践

接下来讲一下Kafka如何做监控,其实只需要上面一幅图,上图绿色指标读,Received是蓝色线,相当于生产者写到Kafka里面,绿色是Spark streaming进行消费,相当于Fetched。这幅图相当于实时同步架构,消息没有做积压。但是为什么波峰会比它高,原因是数据通过Maxwell发送到Kafka时是一个Json数据,但是Kafka消费时需要额外加一些东西(来自哪个topic、offset是什么等),如果两条线没问题就是没出问题。

Bug方面,PHOENIX-4056:java.lang.IllegalArgumentException: Can not create a Path from an empty string,先前有问题采用降版本,目前已经解决,方案在社区里有。SPARK-22968:java.lang.IllegalStateException: No current assignment for partition kssh-2,这个是Sparkstreaming读Kafka时抛出的错误,这个在Spark2.4.0有新的补丁。

接下来分享两个案例,分为两种,一种是3次RIT,园区断电机器挂掉出现RIT。HBase有个WAL,数据基本不会丢,只需要将机器重启。重启过程会有一些RIT操作,如果regionserver挂了申请维护时间,尝试重启regionserver节点,如果不行重启HBase集群,这个时候需要看HBase的master的active的log日志。还有一次是高并发内存不够用,regionserver挂掉,重启后在CDH的HBase运行正常,但是在监控页面HBase还是异常,这时候只需要将CMS的serviceMonitor重启就OK。第三次RIT事故regionserver挂掉,尝试使用HBCK命令修复问题还是很多。最后通过日志分析发现Hlog有问题,通过HDFS命令将文件移到某个地方,重启就OK了。丢失的数据通过Maxwell恢复,预估事故发生点通过全表扫描进行恢复。

回顾·基于HBase实时数仓探索实践

接下来分享一个三支烟的故事,数据来源于阿里云,自建机房需要通过VPN将数据拉倒本地机房。双十一所有仓库都在运作,MySQL机器扛不住导致延迟比较大,延迟约半个小时。需要在T2将数据完全恢复,解决方案直接将Maxwell架设到阿里云进行实时同步,数据进行全表扫描,只需要扫描大屏显示需要的数据,将T1到T2的数据进行SparkSQL,将计算结果写到redis里面Sparkstreaming进行现场改,只判断T2流进的数据才会将T2的基础值进行累积计算,实时Job跑了15分钟数据就实时过来了。

作者介绍:

武基鹏,上海久耶供应链管理有限公司大数据研发工程师。主要从事大数据平台产品的技术工作;负责设计、构建和优化基于HDFS/HBase的存储平台架构;负责提升Hadoop/HBase等集群的高可用性、高性能、高扩展特性;负责基于Spark开发及性能调优。


北京上海深圳杭州Hbase交流群招募中感兴趣的小伙伴,欢迎加管理员微信入群:tianmuqu


——END——

内推信息:

上海久耶供应链招聘大数据工程师,有意者识别下方二维码投递简历。

回顾·基于HBase实时数仓探索实践

文章没看够?下面还有:

彩蛋:武汉HBase MeetUp火热报名进行中。。。


版权声明:本站内容全部来自于腾讯微信公众号,属第三方自助推荐收录。《回顾·基于HBase实时数仓探索实践》的版权归原作者「DataFunTalk」所有,文章言论观点不代表Lambda在线的观点, Lambda在线不承担任何法律责任。如需删除可联系QQ:516101458

文章来源: 阅读原文

相关阅读

举报