使用 Bulk Load 快速向 HBase 中导入数据

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

private static Job createCommitableJob(String tableNameStr, String inputPathStr, String outputPathStr) {
    Configuration conf = HBaseConfiguration.create(new Configuration());
    JobClient client = new JobClient(conf);

    Path inputPath = new Path(inputPathStr);
    Path outputPath = new Path(outputPathStr);

    Job job = Job.getInstance(conf, "load_data_to_" + tableNameStr);
    job.setJarByClass(App.class);

    FileInputFormat.setInputPaths(job, inputPath);
    job.setInputFormatClass(TextInputFormat.class);

    //set mapper class according to job type.
    switch (tableNameStr) {
        case RECORD:
            job.setMapperClass(RecordMapper.class);
            break;
        case XXX:
            job.setMapperClass(XXXMapper.class);
            break;
        case XXXX:
            job.setMapperClass(XXXXMapper.class);
            break;
        default:
            return null;
    }
    job.setMapOutputKeyClass(ImmutableBytesWritable.class);
    job.setMapOutputValueClass(Put.class);
    job.setReducerClass(PutSortReducer.class);

    FileSystem hdfs = FileSystem.newInstance(conf);
    if (null != hdfs) {
        if (hdfs.exists(outputPath)) {
            hdfs.delete(outputPath, true);
        }
    }
    FileOutputFormat.setOutputPath(job, outputPath);

    //for hbase version 1.0.0+
    Connection connection = ConnectionFactory.createConnection(conf);
    TableName tableName = TableName.valueOf(tableNameStr);
    Table table = connection.getTable(tableName);
    HFileOutputFormat2.configureIncrementalLoad(job, table,connection.getRegionLocator(tableName));

    //for hbase 0.96
    /*HTable table = new HTable(conf, tableNameStr);
    HFileOutputFormat.configureIncrementalLoad(job, table);
    TableMapReduceUtil.addDependencyJars(job);
    TableMapReduceUtil.addDependencyJars(job.getConfiguration(), com.google.common.base.Function.class);*/

    return job;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

    //先一些初始化操作，获取作业基本信息，如路径、表名等。
    //创建 MapReduce 作业
    Job job = createCommitableJob(tableNameStr, inputPathStr, outputPathStr);
    if (job == null) {
        LOG.error("Error in create job!");
        return;
    }
    if (job.waitForCompletion(true)) {
        Counter counter = job.getCounters().findCounter(TaskCounter.MAP_OUTPUT_RECORDS);
        LOG.info("job finished, total " + counter.getValue() + " records!");

        //完成 mapreduce 作业后，使用 bulk load导入数据
        Utils.doBulkLoad(conf, hfilePathStr, tableNameStr);
    } else {
        //作业运行失败
        LOG.error("job failed!");
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

public class RecordMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {

    private static final byte[] FAMILY_BYTE = Bytes.toBytes(CommonConfig.HBASE_FAMILY);
    private static final byte[] QUALIFER_BYTE = Bytes.toBytes(CommonConfig.HBASE_QUALIFER);

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //一些准备工作
        //...
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String record = value.toString();
        //过滤无效数据
        if (isVaild()) {
            //数据处理，如格式转换等操作
            //获取RowKey
            byte[] rkValu = getRowKey(record);
            ImmutableBytesWritable rowKey = new ImmutableBytesWritable(rkValue);
            //创建Put对象
            Put put = new Put(rowKey.copyBytes());

            //获取TimeStamp
            long timestamp = getTimeStamp(record);

            //获取应该插入到HBase中的一个cell
            String cellValue = getCellValue(record);

            //将待插入数据存放至Put对象中
            put.add(FAMILY_BYTE, QUALIFER_BYTE, timestamp, Bytes.toBytes(cellValue));
            context.write(rowKey, put);
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        //map完成后的一些清理工作
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

/**
 * loading HFile in 'hFilePath' to HBase, target HTable's name is 'tableNameStr'
 * @param  conf         Configuration instance
 * @param  hFilePath
 * @param  tableNameStr
 * @throws Exception
 */
public static void doBulkLoad(Configuration conf, String hFilePath, String tableNameStr) throws Exception{
    //change permission first.
    FileSystem fs = FileSystem.newInstance(conf);
    chmod(new Path(hFilePath), fs);
    //do bulk load.
    HBaseConfiguration.addHbaseResources(conf);
    LoadIncrementalHFiles loadFiles = new LoadIncrementalHFiles(conf);
    Connection connection = ConnectionFactory.createConnection(conf);
    TableName tableName = TableName.valueOf(tableNameStr);
    Table table = connection.getTable(tableName);
    loadFiles.doBulkLoad(new Path(hFilePath), table);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

/**
 * change the permission of a give path to 777, all subdir are changed recursively.
 * @param path
 * @param fs
 * @throws IOException
 */
public static void chmod(Path path, FileSystem fs) throws IOException {
    fs.setPermission(path, FsPermission.createImmutable(FULL_GRANTS));
    if (fs.getFileStatus(path).isFile()) {
        return;
    }
    RemoteIterator<LocatedFileStatus> fileStatuses = fs.listLocatedStatus(path);
    while(fileStatuses.hasNext()) {
        LocatedFileStatus status = fileStatuses.next();
        if (status != null) {
            fs.setPermission(status.getPath(), FsPermission.createImmutable(FULL_GRANTS));
            chmod(status.getPath(), fs);
        }
    }
}