HBase数据迁移详解！

女爷i 2022-12-20 13:59 94阅读 0赞

### 目录 ###

*   *  一、数据迁移方案
     *  二、数据迁移实施
     *   *  （1）通过importtsv命令
         *  （2）通过importtsv命令+completebulkload
         *  （3）不同文件中数据分割符的处理
         *  （4）自定义MR程序生成HFfile文件（企业常用的方案）

## 一、数据迁移方案 ##

数据迁移，更多的场景是外部的数据源如何将数据写入到HBase  
1.数据库RDBMS

1)sqoop 
    2)kettle ETL工具
    3)其他方式
      **写程序
      **导出文件加载

2.数据文件(log)

1）flume:实时数据收集，将数据的数据插入到HBase
    
      source,channel,sink
    
    2）MapReduce
      input file -> mr  -> hbase table
    
    3）completebulkload（常用）
    
      input file -> mr  -> hfile -> completebulkload -> hbase table

## 二、数据迁移实施 ##

### （1）通过importtsv命令 ###

通过importtsv命令，将文件直接导入到HBase

importtsv -Dimporttsv.columns=a,b,c <tablename> <inputdir>

准备数据：stu.tsv

0001    henry   20      city-1
    0002    cherry  30      city-2
    0003    alex    29      city-3

将数据放在HDFS上

bin/hdfs dfs -put /opt/datas/stu.tsv /user/caizhengjie/datas

export HADOOP_HOME=/opt/modules/hadoop
    export HBASE_HOME=/opt/modules/hbase
    HADOOP_CLASSPATH=`${ HBASE_HOME}/bin/hbase classpath` \
    ${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/lib/hbase-server-1.2.0-cdh5.9.3.jar \
    importtsv \
    -Dimporttsv.columns=HBASE_ROW_KEY,info:username,info:age,info:address \
    stutsv \
    hdfs://bigdata-pro-m01:9000/user/caizhengjie/datas/stu.tsv

运行结果：

hbase(main):004:0> scan 'stutsv'
    ROW                                  COLUMN+CELL                                                                                            
     0001                                column=info:address, timestamp=1605020174889, value=city-1                                             
     0001                                column=info:age, timestamp=1605020174889, value=20                                                     
     0001                                column=info:username, timestamp=1605020174889, value=henry                                             
     0002                                column=info:address, timestamp=1605020174889, value=city-2                                             
     0002                                column=info:age, timestamp=1605020174889, value=30                                                     
     0002                                column=info:username, timestamp=1605020174889, value=cherry                                            
     0003                                column=info:address, timestamp=1605020174889, value=city-3                                             
     0003                                column=info:age, timestamp=1605020174889, value=29                                                     
     0003                                column=info:username, timestamp=1605020174889, value=alex

### （2）通过importtsv命令+completebulkload ###

importtsv  -Dimporttsv.bulk.output=/path/for/output

通过此命令，我们可以将外部的数据文件直接生成一个HFfile文件，然后通过completebulkload直接加载到HBase数据表中

执行流程：

log文件  ->  HFfile文件  ->HBase table表中

**第一步：**

export HADOOP_HOME=/opt/modules/hadoop
    export HBASE_HOME=/opt/modules/hbase
    HADOOP_CLASSPATH=`${ HBASE_HOME}/bin/hbase classpath` \
    ${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/lib/hbase-server-1.2.0-cdh5.9.3.jar \
    importtsv \
    -Dimporttsv.columns=HBASE_ROW_KEY,info:username,info:age,info:address \
    -Dimporttsv.bulk.output=hdfs://bigdata-pro-m01:9000/user/caizhengjie/hfoutput \
    stutsv \
    hdfs://bigdata-pro-m01:9000/user/caizhengjie/datas/stu.tsv

这时在HDFS上会生成HFfile文件，会放在hfoutput里面

**第二步:**

export HADOOP_HOME=/opt/modules/hadoop
    export HBASE_HOME=/opt/modules/hbase
    HADOOP_CLASSPATH=`${ HBASE_HOME}/bin/hbase classpath` \
    ${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/lib/hbase-server-1.2.0-cdh5.9.3.jar completebulkload \
    hdfs://bigdata-pro-m01:9000/user/caizhengjie/hfoutput \
    stutsv

这时HDFS上的HFfile文件会写入到HBase中

运行结果：

hbase(main):003:0> scan 'stutsv'
    ROW                                  COLUMN+CELL                                                                                            
     0001                                column=info:address, timestamp=1605091109579, value=city-1                                             
     0001                                column=info:age, timestamp=1605091109579, value=20                                                     
     0001                                column=info:username, timestamp=1605091109579, value=henry                                             
     0002                                column=info:address, timestamp=1605091109579, value=city-2                                             
     0002                                column=info:age, timestamp=1605091109579, value=30                                                     
     0002                                column=info:username, timestamp=1605091109579, value=cherry                                            
     0003                                column=info:address, timestamp=1605091109579, value=city-3                                             
     0003                                column=info:age, timestamp=1605091109579, value=29                                                     
     0003                                column=info:username, timestamp=1605091109579, value=alex                                              
    3 row(s) in 0.1060 seconds

### （3）不同文件中数据分割符的处理 ###

在实际业务中，我们的数据不可能都是tab键分割开来的，也会出现csv格式的文件，那么我们会根据不同的业务场景，对不同文件中数据分割符做处理。

'-Dimporttsv.separator=|'

准备数据：按照逗号分隔的数据

0001,henry,20,city-1
    0002,cherry,30,city-2
    0003,alex,28,city-3
    0004,lili,35,city-4
    0005,jack,18,city-5

将数据上传到HDFS

bin/hdfs dfs -put /opt/datas/stu.csv /user/caizhengjie/datas

**第一步:生成HFfile文件**

export HADOOP_HOME=/opt/modules/hadoop
    export HBASE_HOME=/opt/modules/hbase
    HADOOP_CLASSPATH=`${ HBASE_HOME}/bin/hbase classpath` \
    ${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/lib/hbase-server-1.2.0-cdh5.9.3.jar \
    importtsv \
    -Dimporttsv.columns=HBASE_ROW_KEY,info:username,info:age,info:address \
    -Dimporttsv.bulk.output=hdfs://bigdata-pro-m01:9000/user/caizhengjie/hfcsv \
    -Dimporttsv.separator=, \
    stutsv \
    hdfs://bigdata-pro-m01:9000/user/caizhengjie/datas/stu.csv

HFfile文件会放在HDFS的`/user/caizhengjie/hfcsv`目录下

**第二步:加载数据，将HFfile文件写入到HBase中**

export HADOOP_HOME=/opt/modules/hadoop
    export HBASE_HOME=/opt/modules/hbase
    HADOOP_CLASSPATH=`${ HBASE_HOME}/bin/hbase classpath` \
    ${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/lib/hbase-server-1.2.0-cdh5.9.3.jar completebulkload \
    hdfs://bigdata-pro-m01:9000/user/caizhengjie/hfcsv \
    stutsv

运行结果：

hbase(main):007:0> scan 'stutsv'
    ROW                                  COLUMN+CELL                                                                                            
     0001                                column=info:address, timestamp=1605096866374, value=city-1                                             
     0001                                column=info:age, timestamp=1605096866374, value=20                                                     
     0001                                column=info:username, timestamp=1605096866374, value=henry                                             
     0002                                column=info:address, timestamp=1605096866374, value=city-2                                             
     0002                                column=info:age, timestamp=1605096866374, value=30                                                     
     0002                                column=info:username, timestamp=1605096866374, value=cherry                                            
     0003                                column=info:address, timestamp=1605096866374, value=city-3                                             
     0003                                column=info:age, timestamp=1605096866374, value=28                                                     
     0003                                column=info:username, timestamp=1605096866374, value=alex                                              
     0004                                column=info:address, timestamp=1605096866374, value=city-4                                             
     0004                                column=info:age, timestamp=1605096866374, value=35                                                     
     0004                                column=info:username, timestamp=1605096866374, value=lili                                              
     0005                                column=info:address, timestamp=1605096866374, value=city-5                                             
     0005                                column=info:age, timestamp=1605096866374, value=18                                                     
     0005                                column=info:username, timestamp=1605096866374, value=jack

### （4）自定义MR程序生成HFfile文件（企业常用的方案） ###

**第一步：编写MR数据迁移程序**

log文件  ->  HFfile文件

package com.kfk.hbase;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/11/10 * @time : 3:31 下午 */
    public class HBaseConstant { 
        public static String HBASE_TABLE = "stu";
        public static String HBASE_CF_INFO = "info";
    }

package com.kfk.hbase;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.hbase.HBaseConfiguration;
    import org.apache.hadoop.hbase.TableName;
    import org.apache.hadoop.hbase.client.*;
    import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
    import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
    import org.apache.hadoop.hbase.util.Bytes;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    import java.io.IOException;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/10/9 * @time : 7:07 下午 */
    
    public class HBaseMRHF extends Configured implements Tool { 
    
        /** * map * TODO */
        public static class MyMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put>{ 
    
            // rowkey,username,age,addres
            // 0001,henry,20,city-1
    
            String[] COLUMN = new String[]{ 
                "rowkey","username","age","addres"
            };
    
            // ImmutableBytesWritable为KEYOUT输出的Key的类型, Put为VALUEOUT输出的Value的类型
            ImmutableBytesWritable rowkey = new ImmutableBytesWritable();
    
            @Override
            public void map(LongWritable key, Text lines, Context context) throws IOException, InterruptedException { 
    
                // 将每一行数据按逗号分开，放入数组中
                String[] values = lines.toString().split(",");
    
                // set rowkey
                rowkey.set(Bytes.toBytes(values[0]));
    
                Put put = new Put(rowkey.get());
    
                for (int index = 1;index < values.length;index++){ 
    
                    put.addImmutable(Bytes.toBytes(HBaseConstant.HBASE_CF_INFO),Bytes.toBytes(COLUMN[index]),Bytes.toBytes(values[index]));
                }
                context.write(rowkey,put);
    
            }
        }
    
    
        /** * run * @param args * @return * @throws IOException * @throws ClassNotFoundException * @throws InterruptedException */
        public int run(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
            // 1) get conf
            Configuration configuration = this.getConf();
    
            // 2) create job
            Job job = Job.getInstance(configuration,this.getClass().getSimpleName());
            job.setJarByClass(this.getClass());
    
            // 3.1) input，指定job的输入
            Path path = new Path(args[0]);
            FileInputFormat.addInputPath(job,path);
    
            // 3.2) map，指定job的mapper和输出的类型
            job.setMapperClass(MyMapper.class);
            job.setMapOutputKeyClass(ImmutableBytesWritable.class);
            job.setMapOutputValueClass(Put.class);
    
    
            // 3.4) output，指定job的输出
            Path outpath = new Path(args[1]);
            FileOutputFormat.setOutputPath(job,outpath);
    
            TableName tableName = TableName.valueOf("stu");
            // 创建一个链接
            Connection connection = ConnectionFactory.createConnection();
            // 获取数据表
            Table table = connection.getTable(tableName);
            RegionLocator regionLocator = connection.getRegionLocator(tableName);
            HFileOutputFormat2.configureIncrementalLoad(job,table,regionLocator);
    
    
            // 4) commit，执行job
            boolean isSuccess = job.waitForCompletion(true);
            // 如果正常执行返回0，否则返回1
            return (isSuccess) ? 0 : 1;
        }
    
        public static void main(String[] args) { 
    
            // 添加输入，输入参数
    // args = new String[]{ 
    // "hdfs://bigdata-pro-m01:9000/user/caizhengjie/datas/stu.csv",
    // "hdfs://bigdata-pro-m01:9000/user/caizhengjie/datas/hfcsv-output"
    // };
    
            Configuration configuration = HBaseConfiguration.create();
    
            try { 
                // 判断输出的文件存不存在，如果存在就将它删除
                Path fileOutPath = new Path(args[1]);
                FileSystem fileSystem = FileSystem.get(configuration);
                if (fileSystem.exists(fileOutPath)){ 
                    fileSystem.delete(fileOutPath,true);
                }
                // 调用run方法
                int status = ToolRunner.run(configuration,new HBaseMRHF(),args);
                // 退出程序
                System.exit(status);
            } catch (IOException e) { 
                e.printStackTrace();
            } catch (ClassNotFoundException e) { 
                e.printStackTrace();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

打包并上传至服务器

**第二步：生成HFfile文件**

export HADOOP_HOME=/opt/modules/hadoop
    export HBASE_HOME=/opt/modules/hbase
    HADOOP_CLASSPATH=`${ HBASE_HOME}/bin/hbase classpath` \
    ${HADOOP_HOME}/bin/hadoop jar /opt/jars/hbase_mrhf.jar \
    hdfs://bigdata-pro-m01:9000/user/caizhengjie/datas/stu.csv hdfs://bigdata-pro-m01:9000/user/caizhengjie/datas/hfcsv-output

**第三步：加载数据，将HFfile文件写入到HBase中**

export HADOOP_HOME=/opt/modules/hadoop
    export HBASE_HOME=/opt/modules/hbase
    HADOOP_CLASSPATH=`${ HBASE_HOME}/bin/hbase classpath` \
    ${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/lib/hbase-server-1.2.0-cdh5.9.3.jar completebulkload \
    hdfs://bigdata-pro-m01:9000/user/caizhengjie/datas/hfcsv-output \
    stutsv

运行结果：

hbase(main):007:0> scan 'stutsv'
    ROW                                     COLUMN+CELL                                                                                                      
     0001                                   column=info:addres, timestamp=1605149649495, value=city-1                                                        
     0001                                   column=info:age, timestamp=1605149649495, value=20                                                               
     0001                                   column=info:username, timestamp=1605149649495, value=henry                                                       
     0002                                   column=info:addres, timestamp=1605149649495, value=city-2                                                        
     0002                                   column=info:age, timestamp=1605149649495, value=30                                                               
     0002                                   column=info:username, timestamp=1605149649495, value=cherry                                                      
     0003                                   column=info:addres, timestamp=1605149649495, value=city-3                                                        
     0003                                   column=info:age, timestamp=1605149649495, value=28                                                               
     0003                                   column=info:username, timestamp=1605149649495, value=alex                                                        
     0004                                   column=info:addres, timestamp=1605149649495, value=city-4                                                        
     0004                                   column=info:age, timestamp=1605149649495, value=35                                                               
     0004                                   column=info:username, timestamp=1605149649495, value=lili                                                        
     0005                                   column=info:addres, timestamp=1605149649495, value=city-5                                                        
     0005                                   column=info:age, timestamp=1605149649495, value=18                                                               
     0005                                   column=info:username, timestamp=1605149649495, value=jack                                                        
    5 row(s) in 0.3760 seconds

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

发表评论取消回复

表情：

评论列表（有 0 条评论，94人围观）

还没有评论，来说两句吧...

相关阅读

相关 hbase 数据迁移： CopyTable

一篇文章介绍了快照方式的迁移： [hbase数据迁移：基于 hbase Snapshot][hbase_ hbase Snapshot]。本文介绍基于CopyTable迁...

淩亂°似流年/ 2024年04月18日 20:43/ 0 赞/ 68 阅读

相关 HBase的数据迁移是如何进行的？

HBase的数据迁移是如何进行的？ HBase是一个高性能的分布式数据库，但在处理大规模数据时，仍然需要进行性能优化以提高查询和写入的效率。下面是一些HBase性能优化的

布满荆棘的人生/ 2024年03月08日 08:31/ 0 赞/ 119 阅读

相关 HBase数据迁移详解！

目录一、数据迁移方案二、数据迁移实施（1）通过importtsv命令（2）通过importts

女爷i/ 2022年12月20日 13:59/ 0 赞/ 95 阅读

相关 Hadoop 数据迁移用法详解

关注我，回复"资料"，获取大数据资料 ![1a21e7c6d575216fe2228a04dc136643.png][] 目录一般用法 hive数据迁移

一时失言乱红尘/ 2022年10月05日 15:52/ 0 赞/ 245 阅读

相关 HBase数据迁移（1）

学习资料：http://www.importnew.com/all-posts HBase数据迁移（1） http://www.importnew.com/3226

心已赠人/ 2022年08月04日 16:38/ 0 赞/ 300 阅读

相关不同集群Hbase数据库数据迁移

不同版本Hbase数据库数据迁移 1、迁出hbase \ 进入hdfs用户 su hdfs \在hdfs上建立test临时目录 hadoop fs

逃离我推掉我的手/ 2022年05月15日 03:41/ 0 赞/ 270 阅读

相关 Hadoop+Hbase集群数据迁移问题

\[size=medium\] 数据迁移或备份是任何一个公司都有可能到遇到的一件事，有关hbase数据迁移，官网也给出了几种方案，这里比较推荐使用hadoop distcp

Dear 丶/ 2022年04月12日 13:18/ 0 赞/ 425 阅读

相关 hbase数据迁移：基于 hbase Snapshot

文章目录欢迎关注本人公众号初始环境说明表迁移手动flush刷新内存数据到磁盘创建快照数据迁移限

た入场券/ 2021年11月18日 01:31/ 0 赞/ 676 阅读

相关 Hive及HBase表的数据迁移

前文：公司最近数据迁移，做了个测试来验证执行命令。一、Hive部分创建原始数据表 create tabl

秒速五厘米/ 2021年11月05日 11:24/ 0 赞/ 426 阅读

相关 HBase的数据迁移及importTsv功能

一、数据迁移场景举例 1. HBase集群装好了，若需要往表中装数据； 2. 数据的备份，如测试集群往生产集群上迁移。二、几种

客官°小女子只卖身不卖艺/ 2021年09月14日 02:58/ 0 赞/ 630 阅读