java - 在hadoop中没有reducer的情况下限制多输出中映射器的数量

Question

您好，我有一个从 HBase 读取记录并将其写入文本文件的应用程序 HBase 表有 200 个区域。我在映射器类中使用 MultipleOutputs 来写入多个文件，并且我正在根据传入的记录创建文件名。

我正在制作 40 个唯一的文件名。我能够正确获取记录，但我的问题是，当 mapreduce 完成时，它会创建 40 个文件以及 2k 个具有正确名称但附加的额外文件用 m-000 等等。

这是因为我有 200 个区域，MultipleOutputs 为每个映射器创建文件，所以 200 个映射器，每个映射器有 40 个唯一文件，所以它创建 40*200 个文件。

如果没有自定义分区程序，我不知道如何避免这种情况。

有什么方法可以强制将记录写入所属文件而不是拆分为多个文件。

我已经使用了自定义分区器类并且它工作正常但我不想使用它因为我只是从 HBase 读取而不是做 reducer 操作。另外如果我必须创建任何额外的文件名那么我必须改变我的代码也是。

这是我的映射器代码

   public class DefaultMapper extends TableMapper<NullWritable, Text> {
        private Text text = new Text();
        MultipleOutputs<NullWritable, Text> multipleOutputs;
        String strName = "";

        @Override()
        public void setup(Context context) throws java.io.IOException, java.lang.InterruptedException {
            multipleOutputs = new MultipleOutputs<NullWritable, Text>(context);
        }
String FILE_NAME = new String(value.getValue(Bytes.toBytes(HbaseBulkLoadMapperConstants.COLUMN_FAMILY),Bytes.toBytes(HbaseBulkLoadMapperConstants.FILE_NAME)));

        multipleOutputs.write(NullWritable.get(), new Text(text.toString()),FILE_NAME);
        //context.write(NullWritable.get(), text);
    }

没有reducer类

这就是我的输出看起来理想情况下应该只创建一个 Japan.BUS.gz 文件。其他文件也是非常小的文件

Japan.BUS-m-00193.gz
Japan.BUS-m-00194.gz
Japan.BUS-m-00195.gz
Japan.BUS-m-00196.gz

Answer 1

我遇到过同样的情况，也做了解决方案。

MultipleOutputs multipleOutputs = null;

String keyToFind = new String();

    public void setup(Context context) throws IOException, InterruptedException
    {
        this.multipleOutputs_normal = new MultipleOutputs<KEYOUT, VALUEOUT>(context);
    }

public void map(NullWritable key , Text values, Context context) throws IOException, InterruptedException
{

   String valToFindInCol[] = values.toString.split(",");/** Lets say comma seperated **/

    if (keyToFind .equals(valToFindInCol[2].toString())|| keyToFind == null) /** Say you need to match 2 position element **/
    {
        this.multipleOutputs.write(NullWritable.get(),<valToWrite>, valToFindInCol[2]);
    } 
    else 
    {
        this.multipleOutputs.close();
        this.multipleOutputs = null;
        this.multipleOutputs = new MultipleOutputs<KEYOUT, VALUEOUT>(context);

        this.multipleOutputs.write(NullWritable.get(),<valToWrite>, valToFindInCol[2]);

    }

    keyToFind=valToFindInCol[2];
}