1. @Override
2. protected void reduce(K row, Iterable<Put> vals, Context context)
3. throws IOException, InterruptedException {
4. long threshold = context.getConfiguration().getLong(
5. "putcombiner.row.threshold", 1L * (1<<30));
6. int cnt = 0;
7. long curSize = 0;
8. Put put = null;
9. Map<byte[], List<Cell>> familyMap = null;
10. for (Put p : vals) {
11. cnt++;
12. if (put == null) {
13. put = p;
14. familyMap = put.getFamilyCellMap();
15. } else {
16. for (Entry<byte[], List<Cell>> entry : p.getFamilyCellMap()
17. .entrySet()) {
18. List<Cell> cells = familyMap.get(entry.getKey());
19. List<Cell> kvs = (cells != null) ? (List<Cell>) cells : null;
20. for (Cell cell : entry.getValue()) {
21. KeyValue kv = KeyValueUtil.ensureKeyValueTypeForMR(cell);
22. curSize += kv.heapSize();
23. if (kvs != null) {
24. kvs.add(kv);
25. }
26. }
27. if (cells == null) {
28. familyMap.put(entry.getKey(), entry.getValue());
29. }
30. }
31. if (cnt % 10 == 0) context.setStatus("Combine " + cnt);
32. if (curSize > threshold) {
33. if (LOG.isDebugEnabled()) {
34. LOG.debug(String.format("Combined %d Put(s) into %d.", cnt, 1));
35. }
36. context.write(row, put);
37. put = null;
38. curSize = 0;
39. cnt = 0;
40. }
41. }
42. }
43. if (put != null) {
44. if (LOG.isDebugEnabled()) {
45. LOG.debug(String.format("Combined %d Put(s) into %d.", cnt, 1));
46. }
47. context.write(row, put);
48. }
49. }

1. public static Job createSubmittableJob(Configuration conf, String[] args)
2. throws IOException {
3. TableName tableName = TableName.valueOf(args[0]);
4. conf.set(TABLE_NAME, tableName.getNameAsString());
5. Path inputDir = new Path(args[1]);
6. Job job = new Job(conf, NAME + "_" + tableName);
7. job.setJarByClass(Importer.class);
8. FileInputFormat.setInputPaths(job, inputDir);
9. job.setInputFormatClass(SequenceFileInputFormat.class);
10. String hfileOutPath = conf.get(BULK_OUTPUT_CONF_KEY);
11. // make sure we get the filter in the jars
12. try {
13. Class<? extends Filter> filter = conf.getClass(FILTER_CLASS_CONF_KEY, null, Filter.class);
14. if (filter != null) {
15. TableMapReduceUtil.addDependencyJarsForClasses(conf, filter);
16. }
17. } catch (Exception e) {
18. throw new IOException(e);
19. }
20. //这里直接写出kv文件。因为数据量大。按region分了
21. if (hfileOutPath != null && conf.getBoolean(HAS_LARGE_RESULT, false)) {
22. LOG.info("Use Large Result!!");
23. try (Connection conn = ConnectionFactory.createConnection(conf);
24. Table table = conn.getTable(tableName);
25. RegionLocator regionLocator = conn.getRegionLocator(tableName)) {
26. HFileOutputFormat2.configureIncrementalLoad(job, table.getTableDescriptor(), regionLocator);
27. job.setMapperClass(KeyValueSortImporter.class);
28. job.setReducerClass(KeyValueReducer.class);
29. Path outputDir = new Path(hfileOutPath);
30. FileOutputFormat.setOutputPath(job, outputDir);
31. job.setMapOutputKeyClass(KeyValueWritableComparable.class);
32. job.setMapOutputValueClass(KeyValue.class);
33. job.getConfiguration().setClass("mapreduce.job.output.key.comparator.class",
34. KeyValueWritableComparable.KeyValueWritableComparator.class,
35. RawComparator.class);
36. Path partitionsPath =
37. new Path(TotalOrderPartitioner.getPartitionFile(job.getConfiguration()));
38. FileSystem fs = FileSystem.get(job.getConfiguration());
39. fs.deleteOnExit(partitionsPath);
40. job.setPartitionerClass(KeyValueWritableComparablePartitioner.class);
41. job.setNumReduceTasks(regionLocator.getStartKeys().length);
42. TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
43. com.google.common.base.Preconditions.class);
44. }
45. //没有分区。
46. } else if (hfileOutPath != null) {
47. job.setMapperClass(KeyValueImporter.class);
48. try (Connection conn = ConnectionFactory.createConnection(conf);
49. Table table = conn.getTable(tableName);
50. RegionLocator regionLocator = conn.getRegionLocator(tableName)){
51. job.setReducerClass(KeyValueSortReducer.class);
52. Path outputDir = new Path(hfileOutPath);
53. FileOutputFormat.setOutputPath(job, outputDir);
54. job.setMapOutputKeyClass(ImmutableBytesWritable.class);
55. job.setMapOutputValueClass(KeyValue.class);
56. HFileOutputFormat2.configureIncrementalLoad(job, table.getTableDescriptor(), regionLocator);
57. TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
58. com.google.common.base.Preconditions.class);
59. }
60. } else {
61. //这个直接调用内TableOutputFarmat。这样就直接调用的是put。这个少量还好，多了不行。
62. //具体的write见下面的代码
63. // No reducers. Just write straight to table. Call initTableReducerJob
64. // because it sets up the TableOutputFormat.
65. job.setMapperClass(Importer.class);
66. TableMapReduceUtil.initTableReducerJob(tableName.getNameAsString(), null, job);
67. job.setNumReduceTasks(0);
68. }
69. return job;
70. }

1. private static KeyValueWritableComparable[] START_KEYS = null;
2. @Override
3. public int getPartition(KeyValueWritableComparable key, KeyValue value,
4. int numPartitions) {
5. for (int i = 0; i < START_KEYS.length; ++i) {
6. if (key.compareTo(START_KEYS[i]) <= 0) {
7. return i;
8. }
9. }
10. return START_KEYS.length;
11. }
12. }

1. @Override
2. public void write(KEY key, Mutation value)
3. throws IOException {
4. if (!(value instanceof Put) && !(value instanceof Delete)) {
5. throw new IOException("Pass a Delete or a Put");
6. }
7. mutator.mutate(value);
8. }

1. public static Job createSubmittableJob(Configuration conf, String[] args)
2. throws IOException {
3. String tableName = args[0];
4. Path outputDir = new Path(args[1]);
5. Job job = new Job(conf, NAME + "_" + tableName);
6. job.setJobName(NAME + "_" + tableName);
7. job.setJarByClass(Export.class);
8. // 定义scan。主要根据配置是否需要设置fitle ，startkey，endkey等。
9.     //简单 Scan s =new Scan（）
10. Scan s = getConfiguredScanForJob(conf, args);
11. //这里会定义每一个region一个map。map的数量等于region的数量。这个map里面基本什么都没做就是读到的
12. //数据直接写出。
13. //这里会定义map的输入格式为TableInputFormat.class
14. IdentityTableMapper.initJob(tableName, s, IdentityTableMapper.class, job);
15. // No reducers. Just write straight to output files.
16. //直接保存数据。
17. job.setNumReduceTasks(0);
18. //输出文件
19. job.setOutputFormatClass(SequenceFileOutputFormat.class);
20. job.setOutputKeyClass(ImmutableBytesWritable.class);
21. job.setOutputValueClass(Result.class);
22. FileOutputFormat.setOutputPath(job, outputDir); // job conf doesn't contain the conf so doesn't have a default fs.
23. return job;
24. }

1. @Override
2. public List<InputSplit> getSplits(JobContext context) throws IOException {
3. List<InputSplit> splits = super.getSplits(context);
4. if ((conf.get(SHUFFLE_MAPS) != null) && "true".equals(conf.get(SHUFFLE_MAPS).toLowerCase())) {
5. Collections.shuffle(splits);
6. }
7. return splits;
8. }

这里getSplits是根据regionLocationInfo ，分区当然是startkey。根据region的数量设置map的个数，这样就可以一个region

对应一个map了。当然这里没有设置，因为没必要。

在初始化map的时候设置了combinerClass为putCombiner

因为hbase 输出都是一个cell单元，如果一行记录包含多个列，就需要这个东西。将相同rowkey的数据放在一块。

对于reduce 根本不需，指定输出格式就行。然后就是位置。

这样Export 过程结束：

2）Import.java

这个刚好相反。需要关注reduce过程

这个主要的就是标红的地方，定义reduce的个数，定义reduce的输出是按region来分区的。这样就ok了。

这里的partition也是按照startkey来区分的