在Hadoop中从映射器生成多个输出文件MapReduce

Question

我正在从mapper生成两个输出文件。我使用Multipleoutput api生成两个输出。我不确定这样做是否正确。这是我的代码。。请仔细检查一下，给我你的建议。。当我运行代码时，我得到了一个错误：java。lang.NullPointerException。。

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.TaskInputOutputContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;


public class Export_Column_Mapping 
{
    public static void main(String[] args) throws Exception 
    {
        String Output_filetype = args[2];
        String Input_column_number = args[3];
        String Output_column_number = args[4];


        Configuration conf = new Configuration();

        conf.setStrings("output_filetype",Output_filetype);
        conf.setStrings("Input_column_number",Input_column_number);
        conf.setStrings("Output_column_number",Output_column_number);

        Job job = new Job(conf, "Export_Column_Mapping");
        job.setJarByClass(Export_Column_Mapping.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        job.setMapperClass(Export_Column_Mapping_Mapper.class);
        job.setNumReduceTasks(0);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }


    public static class Export_Column_Mapping_Mapper extends Mapper<LongWritable,Text,Text,Text>
    {   
        private MultipleOutputs<Text, LongWritable> mos ;

        public void setup(TaskInputOutputContext<?, ?, Text, LongWritable> context) {
            mos = new MultipleOutputs<Text, LongWritable>(context);
        }

        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
        {
            String str_Output_filetype = context.getConfiguration().get("output_filetype"); 

            String str_Input_column_number = context.getConfiguration().get("Input_column_number");
            String[] input_columns_number = str_Input_column_number.split(",");

            String str_Output_column_number= context.getConfiguration().get("Output_column_number");    
            String[] output_columns_number = str_Output_column_number.split(",");

            String str_line = value.toString();
            String[] input_column_array = str_line.split(",");
            String[] Detail_output_column_array = new String[27];
            String[] Shop_output_column_array = new String[8];
            String details_output = null ;
            String Shop_output = null;
        try
        {

            for(int i = 0;i<=input_column_array.length+1; i++)
            {
                int int_outputcolumn = Integer.parseInt(output_columns_number[i]);
                int int_inputcolumn = Integer.parseInt(input_columns_number[i]);

                if((int_inputcolumn != 0) && (int_outputcolumn != 0) && output_columns_number.length == input_columns_number.length){

                    Detail_output_column_array[int_outputcolumn-1] = input_column_array[int_inputcolumn-1];
                    Shop_output_column_array[0] = Detail_output_column_array[0];
                    Shop_output_column_array[1] = Detail_output_column_array[1];
                    Shop_output_column_array[2] = Detail_output_column_array[2];
                    Shop_output_column_array[3] = Detail_output_column_array[3];
                    Shop_output_column_array[4] = Detail_output_column_array[14];


                    if(details_output != null)
                    {
                        details_output = details_output+"       "+ Detail_output_column_array[int_outputcolumn-1];
                        Shop_output = Shop_output+"     "+ Shop_output_column_array[int_outputcolumn-1];

                    }else
                    {
                        details_output = Detail_output_column_array[int_outputcolumn-1];
                        Shop_output =  Shop_output_column_array[int_outputcolumn-1];

                    }
                }
            }

        }catch (Exception e)
        {

        }
            mos.write("Details File", null, details_output);
            mos.write("Shop File", null, Shop_output);

    }
}
}

这是日志...

错误：java。com上的lang.NullPointerException。尼尔森。grfe。Export\u Column\u Mapping$Export\u Column\u Mapping\u Mapper。在com上映射（Export\u Column\u Mapping.java:113）。尼尔森。格菲。Export\u Column\u Mapping$Export\u Column\u Mapping\u Mapper。在组织中映射（Export\u Column\u Mapping.java:1）。阿帕奇。hadoop。mapreduce。映射器。在org上运行（Mapper.java:145）。阿帕奇。hadoop。映射。MapTask。在org上运行NewMapper（MapTask.java:787）。阿帕奇。hadoop。映射。MapTask。在组织上运行（MapTask.java:341）。阿帕奇。hadoop。映射。YarnChild 2美元。在java上运行（YarnChild.java:163）。安全AccessController。javax上的doPrivileged（本机方法）。安全授权。主题doAs（Subject.java:415）位于org。阿帕奇。hadoop。安全用户组信息。doAs（UserGroupInformation.java:1671）位于org。阿帕奇。hadoop。映射。YarnChild。main（YarnChild.java:158）

Answer 1

多输出的定义如下：

private MultipleOutputs<Text, LongWritable> mos ;

同时编写以下是代码：-

mos.write("Details File", null, details_output);
mos.write("Shop File", null, Shop_output);

应该是这样的

mos.write(your key in definition, your value in definition, output directory name as string)

改变你的定义，写和对应的东西如下：

private MultipleOutputs<NullWritable, Text> mos ;
mos.write(NullWritable.get(),new Text(details_output),"Details File");
mos.write(NullWritable.get(),new Text(Shop_output),"Shop File");
 public static class Export_Column_Mapping_Mapper extends Mapper<LongWritable,Text,NullWritable,Text>
{   
job.setOutputKeyClass(NullWritable.class);    

还包括清理代码：

protected void cleanup(Context context)
throws IOException, InterruptedException {
multipleOutputs.close();
}