如何为Hadoop作业配置多个HBase表与文件作为输入？

我之前也碰到过这个棘手的问题——单独用TableMapReduceUtil.initTableMapperJob能处理多HBase表，单独用MultipleInputs能加文件输入，但两者结合就失效。本质是因为默认的TableInputFormat依赖全局配置传递表名和扫描规则，多表场景下会冲突，而initTableMapperJob又会直接覆盖作业的输入格式，导致MultipleInputs的设置被忽略。

下面给你一个可靠的解决方案，核心是为每个HBase表创建独立的可配置TableInputFormat子类，让每个表的参数互不干扰，再通过MultipleInputs统一整合所有输入源：

1. 自定义可配置的TableInputFormat子类 #

我们需要一个能从自定义配置项读取表名和扫描规则的TableInputFormat，这样每个HBase表可以用独立的配置参数：

import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.conf.Configuration;
import java.io.IOException;

public class CustomTableInputFormat extends TableInputFormat {
    @Override
    protected void initialize(Configuration conf) throws IOException {
        // 从自定义配置键中读取当前表的信息
        String tableName = conf.get("hbase.custom.table.name");
        String scanStr = conf.get("hbase.custom.table.scan");
        
        Scan scan = TableMapReduceUtil.convertStringToScan(scanStr);
        setTableName(TableName.valueOf(tableName));
        setScan(scan);
        
        super.initialize(conf);
    }
}

2. 为每个HBase表配置独立的参数并添加到MultipleInputs #

接下来，我们为每个HBase表创建独立的Configuration副本，设置对应的表名和扫描规则，再通过MultipleInputs添加，同时正常添加文件输入：

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.conf.Configuration;

// 假设你已经初始化了基础的Job和Configuration对象
Job job = Job.getInstance(conf, "MultiSourceJob");

// 配置第一个HBase表
Configuration table1Conf = new Configuration(conf);
table1Conf.set("hbase.custom.table.name", "your_first_table");
Scan scan1 = new Scan();
// 这里可以设置scan1的过滤、列族等规则
table1Conf.set("hbase.custom.table.scan", TableMapReduceUtil.convertScanToString(scan1));
// 用虚拟路径区分不同HBase表，比如hbase://table1，TableInputFormat实际不依赖这个路径
MultipleInputs.addInputPath(job, new Path("hbase://table1"), CustomTableInputFormat.class, YourHBaseMapper.class, table1Conf);

// 配置第二个HBase表
Configuration table2Conf = new Configuration(conf);
table2Conf.set("hbase.custom.table.name", "your_second_table");
Scan scan2 = new Scan();
// 设置scan2的规则
table2Conf.set("hbase.custom.table.scan", TableMapReduceUtil.convertScanToString(scan2));
MultipleInputs.addInputPath(job, new Path("hbase://table2"), CustomTableInputFormat.class, YourHBaseMapper.class, table2Conf);

// 添加文件输入，和常规用法一致
MultipleInputs.addInputPath(job, new Path("/path/to/your/file"), TextInputFormat.class, YourFileMapper.class);

// 后续的作业配置（设置Reducer、输出格式等）...
job.waitForCompletion(true);

3. 注意事项 #

• Mapper兼容性：如果HBase输入和文件输入的处理逻辑不同，建议分开写两个Mapper（比如YourHBaseMapper处理HBase的ImmutableBytesWritable和Result，YourFileMapper处理文件的LongWritable和Text），再在MultipleInputs中分别指定。如果逻辑一致，也可以写一个通用Mapper兼容两种输入类型。
• 虚拟路径：HBase输入的路径只是用来区分不同的输入源，CustomTableInputFormat不会实际读取这个路径，所以随便设个唯一的字符串就行，比如hbase://tableX。
• 配置隔离：每个HBase表用独立的Configuration副本，避免参数互相覆盖。

这样就能同时实现多个HBase表和文件作为Hadoop作业的输入了，亲测有效！

内容的提问来源于stack exchange，提问作者AdamSkywalker