HBase BulkLoad批量写入数据实战
1.概述
在进行数据传输中,批量加载数据到HBase集群有多种方式,比如通过HBase API进行批量写入数据、使用Sqoop工具批量导数到HBase集群、使用MapReduce批量导入等。这些方式,在导入数据的过程中,如果数据量过大,可能耗时会比较严重或者占用HBase集群资源较多(如磁盘IO、HBase Handler数等)。今天这篇博客笔者将为大家分享使用HBase BulkLoad的方式来进行海量数据批量写入到HBase集群。
2.内容
在使用BulkLoad之前,我们先来了解一下HBase的存储机制。HBase存储数据其底层使用的是HDFS来作为存储介质,HBase的每一张表对应的HDFS目录上的一个文件夹,文件夹名以HBase表进行命名(如果没有使用命名空间,则默认在default目录下),在表文件夹下存放在若干个Region命名的文件夹,Region文件夹中的每个列簇也是用文件夹进行存储的,每个列簇中存储就是实际的数据,以HFile的形式存在。路径格式如下:
/hbase/data/default/<tbl_name>/<region_id>/<cf>/<hfile_id>
2.1 实现原理
按照HBase存储数据按照HFile格式存储在HDFS的原理,使用MapReduce直接生成HFile格式的数据文件,然后在通过RegionServer将HFile数据文件移动到相应的Region上去。流程如下图所示:
2.2. 生成HFile文件
HFile文件的生成,可以使用MapReduce来进行实现,将数据源准备好,上传到HDFS进行存储,然后在程序中读取HDFS上的数据源,进行自定义封装,组装RowKey,然后将封装后的数据在回写到HDFS上,以HFile的形式存储到HDFS指定的目录中。实现代码如下:
/** * Read DataSource from hdfs & Gemerator hfile. * * @author smartloli. * * Created by Aug 19, 2018 */ public class GemeratorHFile2 { static class HFileImportMapper2 extends Mapper<LongWritable, Text, ImmutableBytesWritable, KeyValue> { protected final String CF_KQ = "cf"; @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); System.out.println("line : " + line); String[] datas = line.split(" "); String row = new Date().getTime() + "_" + datas[1]; ImmutableBytesWritable rowkey = new ImmutableBytesWritable(Bytes.toBytes(row)); KeyValue kv = new KeyValue(Bytes.toBytes(row), this.CF_KQ.getBytes(), datas[1].getBytes(), datas[2].getBytes()); context.write(rowkey, kv); } } public static void main(String[] args) { if (args.length != 1) { System.out.println("<Usage>Please input hbase-site.xml path.</Usage>"); return; } Configuration conf = new Configuration(); conf.addResource(new Path(args[0])); conf.set("hbase.fs.tmp.dir", "partitions_" + UUID.randomUUID()); String tableName = "person"; String input = "hdfs://nna:9000/tmp/person.txt"; String output = "hdfs://nna:9000/tmp/pres"; System.out.println("table : " + tableName); HTable table; try { try { FileSystem fs = FileSystem.get(URI.create(output), conf); fs.delete(new Path(output), true); fs.close(); } catch (IOException e1) { e1.printStackTrace(); } Connection conn = ConnectionFactory.createConnection(conf); table = (HTable) conn.getTable(TableName.valueOf(tableName)); Job job = Job.getInstance(conf); job.setJobName("Generate HFile"); job.setJarByClass(GemeratorHFile2.class); job.setInputFormatClass(TextInputFormat.class); job.setMapperClass(HFileImportMapper2.class); FileInputFormat.setInputPaths(job, input); FileOutputFormat.setOutputPath(job, new Path(output)); HFileOutputFormat2.configureIncrementalLoad(job, table); try { job.waitForCompletion(true); } catch (InterruptedException e) { e.printStackTrace(); } catch (ClassNotFoundException e) { e.printStackTrace(); } } catch (Exception e) { e.printStackTrace(); } } }
在HDFS目录/tmp/person.txt中,准备数据源如下:
1 smartloli 100 2 smartloli2 101 3 smartloli3 102
然后,将上述代码编译打包成jar,上传到Hadoop集群进行执行,执行命令如下:
hadoop jar GemeratorHFile2.jar /data/soft/new/apps/hbaseapp/hbase-site.xml
如果在执行命令的过程中,出现找不到类的异常信息,可能是本地没有加载HBase依赖JAR包,在当前用户中配置如下环境变量信息:
export HADOOP_CLASSPATH=$HBASE_HOME/lib/*:classpath
然后,执行source命令使配置的内容立即生生效。
2.3. 执行预览
在成功提交任务后,Linux控制台会打印执行任务进度,也可以到YARN的资源监控界面查看执行进度,结果如下所示:
等待任务的执行,执行完成后,在对应HDFS路径上会生成相应的HFile数据文件,如下图所示:
2.4 使用BulkLoad导入到HBase
然后,在使用BulkLoad的方式将生成的HFile文件导入到HBase集群中,这里有2种方式。一种是写代码实现导入,另一种是使用HBase命令进行导入。
2.4.1 代码实现导入
通过LoadIncrementalHFiles类来实现导入,具体代码如下:
/** * Use BulkLoad inport hfile from hdfs to hbase. * * @author smartloli. * * Created by Aug 19, 2018 */ public class BulkLoad2HBase { public static void main(String[] args) throws Exception { if (args.length != 1) { System.out.println("<Usage>Please input hbase-site.xml path.</Usage>"); return; } String output = "hdfs://cluster1/tmp/pres"; Configuration conf = new Configuration(); conf.addResource(new Path(args[0])); HTable table = new HTable(conf, "person"); LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf); loader.doBulkLoad(new Path(output), table); } }
执行上述代码,运行结果如下:
2.4.2 使用HBase命令进行导入
先将生成好的HFile文件迁移到目标集群(即HBase集群所在的HDFS上),然后在使用HBase命令进行导入,执行命令如下:
# 先使用distcp迁移hfile hadoop distcp -Dmapreduce.job.queuename=queue_1024_01 -update -skipcrccheck -m 10 /tmp/pres hdfs://nns:9000/tmp/pres # 使用bulkload方式导入数据 hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles /tmp/pres person
最后,我们可以到指定的RegionServer节点上查看导入的日志信息,如下所示为导入成功的日志信息:
2018-08-19 16:30:34,969 INFO [B.defaultRpcServer.handler=7,queue=1,port=16020] regionserver.HStore: Successfully loaded store file hdfs://cluster1/tmp/pres/cf/7b455535f660444695589edf509935e9 into store cf (new location: hdfs://cluster1/hbase/data/default/person/2d7483d4abd6d20acdf16533a3fdf18f/cf/d72c8846327d42e2a00780ac2facf95b_SeqId_4_)
2.5 验证
使用BulkLoad方式导入数据后,可以进入到HBase集群,使用HBase Shell来查看数据是否导入成功,预览结果如下:
3.总结
本篇博客为了演示实战效果,将生成HFile文件和使用BulkLoad方式导入HFile到HBase集群的步骤进行了分解,实际情况中,可以将这两个步骤合并为一个,实现自动化生成与HFile自动导入。如果在执行的过程中出现RpcRetryingCaller的异常,可以到对应RegionServer节点查看日志信息,这里面记录了出现这种异常的详细原因。
4.结束语
这篇博客就和大家分享到这里,如果大家在研究学习的过程当中有什么问题,可以加群进行讨论或发送邮件给我,我会尽我所能为您解答,与君共勉!
另外,博主出书了《Hadoop大数据挖掘从入门到进阶实战》,喜欢的朋友或同学, 可以在公告栏那里点击购买链接购买博主的书进行学习,在此感谢大家的支持。
邮箱:smartloli.org@gmail.com
Twitter: https://twitter.com/smartloli
QQ群(Hadoop - 交流社区1): 424769183
温馨提示:请大家加群的时候写上加群理由(姓名+公司/学校),方便管理员审核,谢谢!
热爱生活,享受编程,与君共勉!
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
Elastic Stack收集Nginx访问日志
一、架构图 image.png 二、搭建 各个组件的软件包,可以到Elastic Stack官网下载 。本次搭建过程使用的是 6.3.1版本。 2.1 安装配置ElasticSearch elasticsearch和logstash都是java开发,运行需要java环境。所以需要先配置jdk tar jdk-8u181-linux-x64.tar.gz -C /opt/ cd /opt && ln -sv jdk1.8.0_181 jdk # 添加环境变量 vim /etc/profile export JAVA_HOME=/opt/jdk export JRE_HOME=/opt/jdk/jre export PATH=$JAVA_HOME/bin:$JRE_HOME:/bin:$PATH source /etc/profile # 解压 [root@monitor elk]# tar xvf elasticsearch-6.3.1.tar.gz [root@monitor elk]# cd elasticsearch-6.3.1/ [root@monitor ela...
- 下一篇
Elasticsearch删除数据之_delete_by_query
es参考版本:elasticsearch:5.5 _delete_by_query会删除所有query语句匹配上的文档,用法如下: curl -X POST "localhost:9200/twitter/_delete_by_query" -H 'Content-Type: application/json' -d' { "query": { "match": { "name": "测试删除" } } } ' 查询必须是有效的键值对,query是键,这和Search API是同样的方式。在search api中q参数和上面效果是一样的。 返回数据格式,告诉你用时和删除多少数据等 { "took" : 147, "timed_out": false, "deleted": 119, "batches": 1, "version_conflicts": 0, "noops": 0, "retries": { "bulk": 0, "search": 0 }, "throttled_millis": 0, "requests_per_second": -1.0, "throttled_un...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
-
Docker使用Oracle官方镜像安装(12C,18C,19C)
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- CentOS8编译安装MySQL8.0.19
- Docker快速安装Oracle11G,搭建oracle11g学习环境
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- MySQL8.0.19开启GTID主从同步CentOS8
- CentOS7,8上快速安装Gitea,搭建Git服务器
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
推荐阅读
最新文章
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- CentOS7编译安装Cmake3.16.3,解决mysql等软件编译问题
- Hadoop3单机部署,实现最简伪集群
- CentOS6,CentOS7官方镜像安装Oracle11G
- CentOS7安装Docker,走上虚拟化容器引擎之路
- SpringBoot2全家桶,快速入门学习开发网站教程
- SpringBoot2初体验,简单认识spring boot2并且搭建基础工程
- Windows10,CentOS7,CentOS8安装MongoDB4.0.16
- CentOS8安装MyCat,轻松搞定数据库的读写分离、垂直分库、水平分库