当前位置: > > > HBase - 数据批量导入教程1(使用MapReduce)

HBase - 数据批量导入教程1(使用MapReduce)

    要实现数据批量导入到 HBase 通常有两种方法:利用 MapReduce 和利用 Bulkload。本文首先介绍前者,即利用 MapReduce 中封装好的方法。在 map 阶段,把数据封装成 Put 操作,直接将数据入库。

一、使用 MapReduce 实现数据批量导入 HBase

1,准备工作

(1)首先我们要准备输入的数据文件 import.dat,其内容如下:
注意:字段分隔符为制表符,建议在服务器上使用 vi 命令创建编辑。
a       c1      name    hangge
a       c1      age     88
b       c1      name    xiaoliu
b       c1      age     19
c       c1      name    lili
c       c1      age     33

(2)然后将该文件传到 HDFS 中:
 hdfs dfs -put import.dat /

(3)接着我们在 HBase 中创建需要的目标表 batch1
create 'batch1','c1'

2,项目配置

(1)项目的 pom.xml 文件中需要添加 hadoop-clienthbase-clienthbase-mapreduce 的依赖:
注意hbase-clienthbase-mapreduce 不能设置 provided,这两个依赖需要打进 jar 包里面,否则会提示找不到对应的类。
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>3.2.0</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-client</artifactId>
    <version>2.3.0</version>
</dependency>
<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-mapreduce</artifactId>
    <version>2.3.0</version>
</dependency>

(2)并且 pom.xml 文件中还需要添加 Maven 的编译打包插件配置:
<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.8.1</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
                <encoding>UTF-8</encoding>
            </configuration>
        </plugin>
        <plugin>
            <artifactId>maven-assembly-plugin</artifactId>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
                <archive>
                    <manifest>
                        <mainClass></mainClass>
                    </manifest>
                </archive>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

3,编写代码

我们编写一个 MapReduce 任务 BatchImportMR,其代码如下,在 map 阶段,把数据封装成 Put 操作,直接将数据入库。
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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 java.io.IOException;

/**
 * 批量导入:
 * 利用 MapReduce 中封装好的方法。
 * 在 map 阶段,把数据封装成 Put 操作,直接将数据入库
 *
 * 注意:需要提前创建表 batch1
 * create 'batch1','c1'
 */
public class BatchImportMR {

    public static class BatchImportMapper extends Mapper<LongWritable, Text, NullWritable, Put>{
        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String[] strs = value.toString().split("\t");
            if(strs.length==4){
                String rowkey = strs[0];
                String columnFamily = strs[1];
                String name = strs[2];
                String val = strs[3];
                Put put = new Put(rowkey.getBytes());
                put.addColumn(columnFamily.getBytes(),name.getBytes(),val.getBytes());
                context.write(NullWritable.get(),put);
            }
        }
    }

    public static void main(String[] args) throws Exception{
        if(args.length!=2){
            //如果传递的参数不够,程序直接退出
            System.exit(100);
        }

        String inPath = args[0];
        String outTableName = args[1];
        //设置属性对应参数
        Configuration conf = new Configuration();
        conf.set("hbase.table.name",outTableName);
        conf.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");

        //封装 Job
        Job job = Job.getInstance(conf, "Batch Import HBase Table:" + outTableName);
        job.setJarByClass(BatchImportMR.class);

        //指定输入路径
        FileInputFormat.setInputPaths(job,new Path(inPath));

        //指定 map 相关的代码
        job.setMapperClass(BatchImportMapper.class);
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(Put.class);

        TableMapReduceUtil.initTableReducerJob(outTableName,null,job);
        TableMapReduceUtil.addDependencyJars(job);

        //禁用 Reduce
        job.setNumReduceTasks(0);

        job.waitForCompletion(true);
    }
}

4,对 MapReduce 任务打 Jar 包

(1)代码编写完毕后,执行打 Jar 包的操作:

(2)打包完毕后,在在项目的 target 目录下看到生成的 XXX-jar-with-dependencies.jar 文件,这个就是我们需要的 jar 包。

5,向集群提交 MapReduce 任务

(1)我们将前面生成的 Jar 包上传至 Hadoop 集群的任意一台机器上,或者 Hadoop 客户端机器上,并且执行如下命令向集群提交任务:
hadoop jar hadoop-0.0.1-SNAPSHOT-jar-with-dependencies.jar BatchImportMR hdfs://node1:9000/import.dat batch1

(2)执行成功之后,查询 HBasebatch1 表中的结果,可以看到数据已经导入成功。
scan 'batch1'
评论0