使用Mapreduce天生Hfile，并通过Bulkload导入到Hbase+源码剖析

• Mapper类：将输入数据转换为HBase的KeyValue格式
  

1. import org.apache.hadoop.hbase.KeyValue;
2. import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
3. import org.apache.hadoop.hbase.util.Bytes;
4. import org.apache.hadoop.io.LongWritable;
5. import org.apache.hadoop.io.Text;
6. import org.apache.hadoop.mapreduce.Mapper;
8. import java.io.IOException;
10. public class HFileMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, KeyValue> {
11.    
12.     private byte[] columnFamily;
13.     private byte[] columnQualifier;
14.     private ImmutableBytesWritable rowKey = new ImmutableBytesWritable();
16.     @Override
17.     protected void setup(Context context) {
18.         // 配置列族和列名（根据实际情况修改）
19.         columnFamily = Bytes.toBytes("cf");
20.         columnQualifier = Bytes.toBytes("col");
21.     }
23.     @Override
24.     protected void map(LongWritable key, Text value, Context context)
25.             throws IOException, InterruptedException {
26.         
27.         String[] fields = value.toString().split(","); // 假设输入格式为: rowkey,value
28.         if (fields.length < 2) return;
30.         String row = fields[0];
31.         String val = fields[1];
32.         
33.         rowKey.set(Bytes.toBytes(row));
34.         
35.         // 创建KeyValue对象（时间戳可自定义或自动生成）
36.         KeyValue kv = new KeyValue(
37.             Bytes.toBytes(row),    // Row key
38.             columnFamily,          // Column family
39.             columnQualifier,       // Column qualifier
40.             System.currentTimeMillis(), // Timestamp
41.             Bytes.toBytes(val)     // Value
42.         );
43.         
44.         context.write(rowKey, kv);
45.     }
46. }

复制代码

•   Driver类：配置MapReduce作业
  

1. import org.apache.hadoop.conf.Configuration;
2. import org.apache.hadoop.fs.Path;
3. import org.apache.hadoop.hbase.HBaseConfiguration;
4. import org.apache.hadoop.hbase.TableName;
5. import org.apache.hadoop.hbase.client.Connection;
6. import org.apache.hadoop.hbase.client.ConnectionFactory;
7. import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
8. import org.apache.hadoop.mapreduce.Job;
9. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
10. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
11. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
13. public class HFileGenerator {
15.     public static void main(String[] args) throws Exception {
16.         Configuration conf = HBaseConfiguration.create();
17.         // 设置HBase ZooKeeper地址（根据实际情况修改）
18.         conf.set("hbase.zookeeper.quorum", "zk1.example.com,zk2.example.com");
19.         
20.         Job job = Job.getInstance(conf, "Generate HFile for HBase");
21.         job.setJarByClass(HFileGenerator.class);
22.         
23.         // 设置Mapper
24.         job.setMapperClass(HFileMapper.class);
25.         job.setMapOutputKeyClass(ImmutableBytesWritable.class);
26.         job.setMapOutputValueClass(KeyValue.class);
27.         
28.         // 设置输入格式和路径
29.         job.setInputFormatClass(TextInputFormat.class);
30.         FileInputFormat.addInputPath(job, new Path(args[0]));  // 输入路径
31.         
32.         // 设置输出格式
33.         job.setOutputFormatClass(HFileOutputFormat2.class);
34.         FileOutputFormat.setOutputPath(job, new Path(args[1])); // HFile输出路径
35.         
36.         // 获取HBase表信息
37.         Connection conn = ConnectionFactory.createConnection(conf);
38.         TableName tableName = TableName.valueOf("your_table_name"); // 目标表名
39.         
40.         // 配置HFileOutputFormat
41.         HFileOutputFormat2.configureIncrementalLoad(
42.             job,
43.             conn.getTable(tableName),
44.             conn.getRegionLocator(tableName)
45.         );
46.         
47.         System.exit(job.waitForCompletion(true) ? 0 : 1);
48.     }
49. }

复制代码

这里重点看一下org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2.class源码，打开一看有一个解释：
  

1. /**
2. * Writes HFiles. Passed Cells must arrive in order. Writes current time as the sequence id for the
3. * file. Sets the major compacted attribute on created {@link HFile}s. Calling write(null,null) will
4. * forcibly roll all HFiles being written.
5. * <p>
6. * Using this class as part of a MapReduce job is best done using
7. * {@link #configureIncrementalLoad(Job, TableDescriptor, RegionLocator)}.
8. */

复制代码

• 写入 HFiles，传入的 Cell 必须按顺序到达。将当前时间作为文件的序列 ID 写入。在创建的 HFile 上设置“主要压缩完成”属性。调用 write(null, null) 将逼迫滚动所有正在写入的 HFiles。 （说明写入前需要排好序。！！！！！）
  

  思考：需不需要手动设置ReduceTask？

不需要，HFileOutputFormat2.configureIncrementalLoad()会根据MapTask输出的Value范例，自动设置Reducer。
   支持的Value范例有：KeyValue.class、MapReduceExtendedCell.class、Put.class、Text.class。
  别的，还会根据表的region情况，设置reduce数目和Partitioner分区器

1. static void configureIncrementalLoad(Job job, List<TableInfo> multiTableInfo,
2.     Class<? extends OutputFormat<?, ?>> cls) throws IOException {
4.     ...........
6.     // Based on the configured map output class, set the correct reducer to properly
7.     // sort the incoming values.
8.     // TODO it would be nice to pick one or the other of these formats.
9.     if (
10.       KeyValue.class.equals(job.getMapOutputValueClass())
11.         || MapReduceExtendedCell.class.equals(job.getMapOutputValueClass())
12.     ) {
13.       job.setReducerClass(CellSortReducer.class);
14.     } else if (Put.class.equals(job.getMapOutputValueClass())) {
15.       job.setReducerClass(PutSortReducer.class);
16.     } else if (Text.class.equals(job.getMapOutputValueClass())) {
17.       job.setReducerClass(TextSortReducer.class);
18.     } else {
19.       LOG.warn("Unknown map output value type:" + job.getMapOutputValueClass());
20.     }
22.     ...........
24.     List<ImmutableBytesWritable> startKeys =
25.       getRegionStartKeys(regionLocators, writeMultipleTables);
26.     // Use table's region boundaries for TOP split points.
27.     LOG.info("Configuring " + startKeys.size() + " reduce partitions "
28.       + "to match current region count for all tables");
29.     job.setNumReduceTasks(startKeys.size());
31.     configurePartitioner(job, startKeys, writeMultipleTables);
33.     ...........
35. }

复制代码

初始化Partitioner: 

1. /**
2.    * Configure <code>job</code> with a TotalOrderPartitioner, partitioning against
3.    * <code>splitPoints</code>. Cleans up the partitions file after job exists.
4.    */
5.   static void configurePartitioner(Job job, List<ImmutableBytesWritable> splitPoints,
6.     boolean writeMultipleTables) throws IOException {
7.     Configuration conf = job.getConfiguration();
8.     // create the partitions file
9.     FileSystem fs = FileSystem.get(conf);
10.     String hbaseTmpFsDir =
11.       conf.get(HConstants.TEMPORARY_FS_DIRECTORY_KEY, HConstants.DEFAULT_TEMPORARY_HDFS_DIRECTORY);
12.     Path partitionsPath = new Path(hbaseTmpFsDir, "partitions_" + UUID.randomUUID());
13.     fs.makeQualified(partitionsPath);
14.     writePartitions(conf, partitionsPath, splitPoints, writeMultipleTables);
15.     fs.deleteOnExit(partitionsPath);
17.     // configure job to use it
18.     job.setPartitionerClass(TotalOrderPartitioner.class);
19.     TotalOrderPartitioner.setPartitionFile(conf, partitionsPath);
20.   }

复制代码

分区器：org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner

（源码太长了，就不贴了）
可以实现全局有序分区，getPartition()方法用于确定给定键值对属于哪个分区。它依靠于之前构建的分区数据结构（Node），在差别情况下有差别的处理处罚逻辑：
1. 键范例为BinaryComparable且启用了自然顺序

如果键范例是BinaryComparable且配置启用了自然顺序，分区器会构建一个前缀树（Trie）。这种情况下，getPartition()方法通过前缀树来查找分区。前缀树的构建基于键的字节体现，前缀树的每个节点体现一个字节，键的字节序列决定了它在前缀树中的路径。这种方法利用了键的二进制体现的自然顺序，可以大概快速定位到正确的分区。前缀树的深度由配置参数total.order.partitioner.max.trie.depth控制，默以为200。
2. 键范例不是BinaryComparable或者自然顺序被禁用

如果键范例不是BinaryComparable或者自然顺序被禁用，分区器会使用二分查找来定位分区。这种情况下，getPartition()方法会调用BinarySearchNode类的findPartition()方法，使用二分查找算法在分区键集合中查找给定键。二分查找的时间复杂度为O(logn)，其中n是分区键的数目。

参考文章：Hbase Bulkload 原理｜口试必备 - 知乎

https://zhuanlan.zhihu.com/p/363044046

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

继续阅读请点击广告