面试题：HBase MapReduce自定义分布式计算的设计与实践

1. 数据输入输出格式的选择

输入格式：

由于数据存储在HBase中，通常选择TableInputFormat作为输入格式。它可以方便地从HBase表中读取数据。在配置TableInputFormat时，需要指定要读取的HBase表名、列族以及所需的列等信息。例如：

job.setInputFormatClass(TableInputFormat.class);
Scan scan = new Scan();
scan.addFamily(Bytes.toBytes("cf")); // 假设列族为cf
scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("col")); // 假设列名为col
TableMapReduceUtil.initTableMapperJob(
    "user_behavior_table", // HBase表名
    scan,
    UserBehaviorMapper.class,
    Text.class,
    IntWritable.class,
    job);

输出格式：
- 如果最终结果还是要存储回HBase，可以选择TableOutputFormat。它能够将MapReduce的输出结果写入到HBase表中。需要配置输出表的相关信息，比如表名等。示例：
```
job.setOutputFormatClass(TableOutputFormat.class);
job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, "user_profile_table");
```
- 如果希望输出为文件（如文本文件等），可以选择TextOutputFormat，将结果以文本形式输出到指定的文件路径。配置如下：
```
job.setOutputFormatClass(TextOutputFormat.class);
FileOutputFormat.setOutputPath(job, new Path("output_path"));
```

2. Mapper和Reducer的逻辑设计

Mapper逻辑：

功能：从HBase表中读取用户行为数据，并进行初步的处理和转换，为后续的Reducer提供合适的输入。

实现：

从TableInputFormat提供的输入中获取每行数据，每行数据是一个Result对象，包含了HBase表中的一行记录。
解析Result对象，提取出所需的用户行为数据字段，如用户ID、行为类型、时间等。例如：

public class UserBehaviorMapper extends TableMapper<Text, IntWritable> {
    @Override
    protected void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException {
        byte[] userIdBytes = value.getValue(Bytes.toBytes("cf"), Bytes.toBytes("user_id"));
        byte[] behaviorTypeBytes = value.getValue(Bytes.toBytes("cf"), Bytes.toBytes("behavior_type"));
        String userId = Bytes.toString(userIdBytes);
        int behaviorType = Bytes.toInt(behaviorTypeBytes);
        context.write(new Text(userId), new IntWritable(behaviorType));
    }
}

Reducer逻辑：

功能：根据Mapper输出的中间结果，进行聚合和计算，构建用户画像。

实现：

接收Mapper输出的以用户ID为键，行为类型相关数据为值的键值对。
对同一用户ID的所有行为数据进行统计和分析。例如，统计每种行为类型的出现次数，从而构建用户的行为画像。示例代码如下：

public class UserProfileReducer extends Reducer<Text, IntWritable, Text, Text> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int clickCount = 0;
        int purchaseCount = 0;
        for (IntWritable value : values) {
            int behaviorType = value.get();
            if (behaviorType == 1) {
                clickCount++;
            } else if (behaviorType == 2) {
                purchaseCount++;
            }
        }
        StringBuilder profileBuilder = new StringBuilder();
        profileBuilder.append("Click Count: ").append(clickCount).append(", Purchase Count: ").append(purchaseCount);
        context.write(key, new Text(profileBuilder.toString()));
    }
}

3. 处理数据倾斜问题

数据采样：
- 在MapReduce任务启动前，对HBase中的数据进行采样。可以通过随机抽取一定比例的行键，分析这些行键的分布情况，找出可能导致数据倾斜的热点区域。例如，可以使用HBase Sampler工具来进行采样。
预分区：
- 根据采样结果，对数据进行预分区。如果发现某些行键范围的数据量特别大，可以在HBase表创建时，将这些行键范围预先划分到不同的Region中，避免数据集中在少数Region上。在HBase Shell中可以使用create 'table_name', {NAME => 'cf', SPLITS => ['split1','split2',...]}命令进行预分区，其中split1、split2等是根据采样结果确定的行键分割点。

Mapper端处理：

在Mapper中，可以对可能导致倾斜的键进行随机化处理。例如，对于某些高频出现的用户ID，可以在其前面添加一个随机前缀，使得这些键分散到不同的Reducer中进行处理。处理后在Reducer端再去掉随机前缀进行聚合。示例代码如下：

public class UserBehaviorMapper extends TableMapper<Text, IntWritable> {
    @Override
    protected void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException {
        byte[] userIdBytes = value.getValue(Bytes.toBytes("cf"), Bytes.toBytes("user_id"));
        byte[] behaviorTypeBytes = value.getValue(Bytes.toBytes("cf"), Bytes.toBytes("behavior_type"));
        String userId = Bytes.toString(userIdBytes);
        int behaviorType = Bytes.toInt(behaviorTypeBytes);
        // 随机化处理
        Random random = new Random();
        int randomPrefix = random.nextInt(10);
        context.write(new Text(randomPrefix + "_" + userId), new IntWritable(behaviorType));
    }
}

在Reducer中：

public class UserProfileReducer extends Reducer<Text, IntWritable, Text, Text> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        String userId = key.toString().split("_")[1];
        int clickCount = 0;
        int purchaseCount = 0;
        for (IntWritable value : values) {
            int behaviorType = value.get();
            if (behaviorType == 1) {
                clickCount++;
            } else if (behaviorType == 2) {
                purchaseCount++;
            }
        }
        StringBuilder profileBuilder = new StringBuilder();
        profileBuilder.append("Click Count: ").append(clickCount).append(", Purchase Count: ").append(purchaseCount);
        context.write(new Text(userId), new Text(profileBuilder.toString()));
    }
}

Combiner使用：

在Mapper和Reducer之间添加Combiner。Combiner的逻辑与Reducer类似，它在Mapper端对局部数据进行初步聚合，减少Mapper到Reducer之间的数据传输量。这对于缓解数据倾斜有一定帮助，因为可以减少在Reducer端需要处理的数据量。例如，可以将上述Reducer的部分逻辑封装为Combiner：

public class UserBehaviorCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int clickCount = 0;
        int purchaseCount = 0;
        for (IntWritable value : values) {
            int behaviorType = value.get();
            if (behaviorType == 1) {
                clickCount++;
            } else if (behaviorType == 2) {
                purchaseCount++;
            }
        }
        context.write(key, new IntWritable(clickCount + purchaseCount));
    }
}

然后在MapReduce作业中设置Combiner：

job.setCombinerClass(UserBehaviorCombiner.class);

星途面试题库

面试题：HBase MapReduce自定义分布式计算的设计与实践

知识考点

面试题答案

1. 数据输入输出格式的选择

2. Mapper和Reducer的逻辑设计

3. 处理数据倾斜问题