为什么 Map record output=0 ，即使我在 mapper 中给出输出答案

【问题标题】：Why Map record output=0 , even when I am giving output in mapper为什么 Map record output=0 ，即使我在 mapper 中给出输出
【发布时间】：2017-09-02 23:47:29
【问题描述】：

我已经尝试了很多，但不明白我的映射器记录输出 =0 的原因。我希望我的映射器在处理大数据时不止一次地读取这些行，并且不止一次需要每行上的数据，所以我首先尝试使用包含 --

的小文件（graph.txt）

   1,2,4,6
   2,10,3,7
   3,6,5,8
   4,7,7,9
   5,13,9,9

但是由于 mapper 逐行处理文件，所以没有其他方法可以在第一次调用 map() 方法 (n-1) 时将所有值存储在文件中，然后在最后一个 map( ) 方法调用。对于文件中的每一行，我将其数据存储在行数组中。并在最后一个 map 方法调用中通过 output.collect() 函数提供输出。另外我正在使用 setup() 方法来计算否。文件中的行数，因为 setup() 为每个映射器调用一次。这里由于输入文件很小，所以只调用了 1 个映射器。

我被困了一段时间，我是新手，请给出一些解决方案。提前致谢。这是代码。

驱动程序代码-

    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapred.FileInputFormat;
    import org.apache.hadoop.mapred.FileOutputFormat;
    import org.apache.hadoop.mapred.JobClient;
    import org.apache.hadoop.mapred.JobConf;
    import org.apache.hadoop.mapred.TextInputFormat;
    import org.apache.hadoop.mapred.TextOutputFormat;


    public class primdriver {
    public static void main(String[] args) throws Exception {
        JobConf conf = new JobConf(primdriver.class);
        conf.setJobName("primdriver");

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

        conf.setMapperClass(primmapper.class);
        //conf.setCombinerClass(Reduce.class);
        conf.setReducerClass(primreducer.class);

        conf.setInputFormat(TextInputFormat.class);
        conf.setOutputFormat(TextOutputFormat.class);

        FileInputFormat.setInputPaths(conf, new Path(args[0]));
        FileOutputFormat.setOutputPath(conf, new Path(args[1]));

       JobClient.runJob(conf);
      }
}

映射器代码 -

    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;

    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapred.MapReduceBase;
    import org.apache.hadoop.mapred.Mapper;
    import org.apache.hadoop.mapred.OutputCollector;
    import org.apache.hadoop.mapred.Reporter;
    import org.apache.hadoop.mapreduce.Mapper.Context;

    public class primmapper extends MapReduceBase implements         
    Mapper<LongWritable, Text, Text, Text> {
        //private final static IntWritable one = new IntWritable(1);
        //private Text word = new Text();
        private int no_line=0;
        private int i=0;
        public void setup(Context context) throws IOException{
            Path pt=new Path("hdfs:/myinput/graph.txt");//Location of file         
            in HDFS
            FileSystem fs = FileSystem.get(new Configuration());
            BufferedReader br=new BufferedReader(new                 
            InputStreamReader(fs.open(pt)));
            String line;
            line=br.readLine();
            while (line != null){
                no_line=no_line+1;
                line=br.readLine();
            }
        }   
        private String [][]row=new String[no_line][4];

        @Override       
        public void map(LongWritable key, Text value, OutputCollector<Text, 
   Text> output, Reporter reporter) throws IOException {        

            if (i<no_line-1){

                String[] s = value.toString().split(",");
                for (int j=0;j<s.length;j++){
                    row[i][j]=(s[j]);
                }
                i=i+1;
            }
            else{
                String[] s = value.toString().split(",");
                for (int j=0;j<s.length;j++){
            //row[i][j]=Integer.parseInt(s[j]);
                }
                for (int i=0;i<no_line-1;i++){
                    String a=row[i][0];
                    String b=row[i][1]+","+row[i][2]+","+row[i][3];
                    output.collect(new Text(a),new Text(b));
                }
            }
        }
    }

Reducer 代码 -

    import java.io.IOException;
    import java.util.Iterator;

    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapred.MapReduceBase;
    import org.apache.hadoop.mapred.OutputCollector;
    import org.apache.hadoop.mapred.Reducer;
    import org.apache.hadoop.mapred.Reporter;


        public class primreducer extends MapReduceBase implements 
        Reducer<Text, Text, Text, Text> {
            public void reduce(Text key, Iterator<Text> values, 
   OutputCollector<Text, Text> output, Reporter reporter) throws IOException         
   {
        int a = 0, b = 0 , c = 0;
        output.collect(new Text("kishan "), new Text("sharma"));
        while (values.hasNext()) {
            String val[]=(values.next().toString()).split(",");
            a=Integer.parseInt(val[0]);
            b=Integer.parseInt(val[1]);
            c=Integer.parseInt(val[2]);
        }
        output.collect(key, new Text(a+","+b+","+c));
    }
}

在控制台中我得到了这些日志-

    [training@localhost workspace]$ hadoop jar hierarchical.jar primdriver 
    myinput/graph.txt cluster5
    17/04/07 10:21:18 WARN mapred.JobClient: Use GenericOptionsParser for         
    parsing the arguments. Applications should implement Tool for the same.
    17/04/07 10:21:18 WARN snappy.LoadSnappy: Snappy native library is available
    17/04/07 10:21:18 INFO snappy.LoadSnappy: Snappy native library loaded
    17/04/07 10:21:18 INFO mapred.FileInputFormat: Total input paths to process : 1
    17/04/07 10:21:18 INFO mapred.JobClient: Running job: job_201704070816_0007
    17/04/07 10:21:19 INFO mapred.JobClient:  map 0% reduce 0%
    17/04/07 10:22:21 INFO mapred.JobClient:  map 100% reduce 0%
    17/04/07 10:22:29 INFO mapred.JobClient:  map 100% reduce 66%
    17/04/07 10:22:53 INFO mapred.JobClient:  map 100% reduce 100%
    17/04/07 10:23:22 INFO mapred.JobClient: Job complete: job_201704070816_0007
    17/04/07 10:23:22 INFO mapred.JobClient: Counters: 33
    17/04/07 10:23:22 INFO mapred.JobClient:   File System Counters
    17/04/07 10:23:22 INFO mapred.JobClient:     FILE: Number of bytes read=6
    17/04/07 10:23:22 INFO mapred.JobClient:     FILE: Number of bytes written=361924
    17/04/07 10:23:22 INFO mapred.JobClient:     FILE: Number of read operations=0
    17/04/07 10:23:22 INFO mapred.JobClient:     FILE: Number of large read operations=0
    17/04/07 10:23:22 INFO mapred.JobClient:     FILE: Number of write operations=0
    17/04/07 10:23:22 INFO mapred.JobClient:     HDFS: Number of bytes read=146
    17/04/07 10:23:22 INFO mapred.JobClient:     HDFS: Number of bytes written=0
    17/04/07 10:23:22 INFO mapred.JobClient:     HDFS: Number of read operations=3
    17/04/07 10:23:22 INFO mapred.JobClient:     HDFS: Number of large read operations=0
    17/04/07 10:23:22 INFO mapred.JobClient:     HDFS: Number of write operations=2
    17/04/07 10:23:22 INFO mapred.JobClient:   Job Counters 
    17/04/07 10:23:22 INFO mapred.JobClient:     Launched map tasks=1
    17/04/07 10:23:22 INFO mapred.JobClient:     Launched reduce tasks=1
    17/04/07 10:23:22 INFO mapred.JobClient:     Data-local map tasks=1
    17/04/07 10:23:22 INFO mapred.JobClient:     Total time spent by all maps in occupied slots (ms)=90240
    17/04/07 10:23:22 INFO mapred.JobClient:     Total time spent by all reduces in occupied slots (ms)=31777
    17/04/07 10:23:22 INFO mapred.JobClient:     Total time spent by all maps waiting after reserving slots (ms)=0
    17/04/07 10:23:22 INFO mapred.JobClient:     Total time spent by all reduces waiting after reserving slots (ms)=0
    17/04/07 10:23:22 INFO mapred.JobClient:   Map-Reduce Framework
    17/04/07 10:23:22 INFO mapred.JobClient:     Map input records=5
    17/04/07 10:23:22 INFO mapred.JobClient:     Map output records=0
    17/04/07 10:23:22 INFO mapred.JobClient:     Map output bytes=0
    17/04/07 10:23:22 INFO mapred.JobClient:     Input split bytes=104
    17/04/07 10:23:22 INFO mapred.JobClient:     Combine input records=0
    17/04/07 10:23:22 INFO mapred.JobClient:     Combine output records=0
    17/04/07 10:23:22 INFO mapred.JobClient:     Reduce input groups=0
    17/04/07 10:23:22 INFO mapred.JobClient:     Reduce shuffle bytes=6
    17/04/07 10:23:22 INFO mapred.JobClient:     Reduce input records=0
    17/04/07 10:23:22 INFO mapred.JobClient:     Reduce output records=0
    17/04/07 10:23:22 INFO mapred.JobClient:     Spilled Records=0
    17/04/07 10:23:22 INFO mapred.JobClient:     CPU time spent (ms)=1240
    17/04/07 10:23:22 INFO mapred.JobClient:     Physical memory (bytes) snapshot=196472832
    17/04/07 10:23:22 INFO mapred.JobClient:     Virtual memory (bytes) snapshot=775897088
    17/04/07 10:23:22 INFO mapred.JobClient:     Total committed heap usage (bytes)=177016832
    17/04/07 10:23:22 INFO mapred.JobClient:   org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter
    17/04/07 10:23:22 INFO mapred.JobClient:     BYTES_READ=42

【问题讨论】：

标签： java hadoop bigdata

【解决方案1】：

当i 小于no_line-1 时，您不会收集任何东西。这种情况始终适用于您的情况，这就是您看不到任何地图输出记录的原因。

当您开始处理第一条记录时，no_line 已经初始化为其最终值（输入文件“hdfs:/myinput/graph.txt”中的实际行数）。

此时，i 为 0。然后，当满足此 if 条件时，i 在此特定映射器中变为 1（根本不是映射器）。* 然后，@987654326 @ 的值为 1（在此映射器中），它必须仍小于 no_line - 1。您的文件graph.txt 似乎超过 5 行（我猜）。

总而言之，setup() 在每个 map() 被每个映射器执行之前执行一次。

我不知道你想做什么，从这部分似乎很难理解。如果您需要更多帮助，请尝试使其更清晰，并使用更多详细信息更新您的问题。在 else 语句中，再次使用变量 i 似乎很混乱，因为不清楚您是否真的想使用本地 i 或“阴影”i。您的 IDE 没有对此发出警告吗？

*这是一种非常糟糕的做法，因为您无法知道i 将在每个映射器中采用哪些值，这取决于数据分区。

【讨论】：

我现在已经澄清了这个问题。我希望现在很清楚，请告诉我这有什么问题。
我已经找到了问题所在。实际上，在调用 setup() 方法后 i 和 no_line 的值都为零。所以它从不执行 if 条件。这就是为什么映射器没有输出的原因。你能告诉我为什么设置不起作用。
@KISHANSHARMA 因为此路径中没有此类文件（检查记录器错误消息）或此文件为空。这并不意味着不检查 if 语句。这意味着它已被检查，但流程进入 else 条件，然后什么也不做，因为 no_line 为 0