Hadoop Mapreduce作业卡在地图100％减少51％

Question

所以，我正在寻找一个无限循环的地方，我不知道是否还有其他任何可以导致这种情况。我正在使用四个群集节点，所以我非常肯定不会缺少RAM，正如其他相同类型的问题所暗示的那样。

我的代码：

package org.myorg; 

import java.io.IOException; 
import java.util.*; 

import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.conf.*; 
import org.apache.hadoop.io.*; 
import org.apache.hadoop.mapreduce.*; 
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.TextOutputFormat; 

import util.hashing.*; 



public class LatLong { 


public static class Map extends Mapper<Object, Text, Text, Text> { 
    //private final static IntWritable one = new IntWritable(1); 


    public void map(Object key, Text value, Context context) throws IOException, InterruptedException { 
     String line = value.toString(); 
     String[] longLatArray = line.split(","); 
     double longi = Double.parseDouble(longLatArray[0]); 
     double lat = Double.parseDouble(longLatArray[1]); 
     //List<Double> origLatLong = new ArrayList<Double>(2); 
     //origLatLong.add(lat); 
     //origLatLong.add(longi); 
     Geohash inst = Geohash.getInstance(); 
     //encode is the library's encoding function 
     String hash = inst.encode(lat,longi); 
     //Using the first 5 characters just for testing purposes 
     //Need to find the right one later 
     int accuracy = 4; 
     //hash of the thing is shortened to whatever I figure out 
     //to be the right size of each tile 
     Text shortenedHash = new Text(hash.substring(0,accuracy)); 
     Text origHash = new Text(hash); 

     context.write(shortenedHash, origHash); 
    } 
} 

public static class Reduce extends Reducer<Text, Text, Text, Text> { 

    private IntWritable totalTileElementCount = new IntWritable(); 
    private Text latlongimag = new Text(); 
    private Text dataSeparator = new Text(); 

    @Override 
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { 
     int elementCount = 0; 
     boolean first = true; 
     Iterator<Text> it = values.iterator(); 
     String lat = new String(); 
     String longi = new String(); 
     Geohash inst = Geohash.getInstance(); 

     while (it.hasNext()) { 
     elementCount = elementCount+1; 
     if(first) 
     { 
      double[] doubleArray = (inst.decode(it.next().toString())); 
      lat = Double.toString(doubleArray[0]); 
      longi = Double.toString(doubleArray[1]); 
      first = false; 

     } 



     } 
     totalTileElementCount.set(elementCount); 
     //Geohash inst = Geohash.getInstance(); 

     String mag = totalTileElementCount.toString(); 

     latlongimag.set(lat+","+ longi +","+mag+","); 
     dataSeparator.set(""); 
     context.write(latlongimag, dataSeparator); 
    } 
} 

public static void main(String[] args) throws Exception { 
    Configuration conf = new Configuration(); 
    Job job = new Job(conf, "wordcount"); 
    job.setJarByClass(LatLong.class); 

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

    job.setMapperClass(Map.class); 
    job.setReducerClass(Reduce.class); 

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

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

    job.waitForCompletion(true); 
} 

}  

Answer 1

内

while (it.hasNext()) { 
     elementCount = elementCount+1; 
     if(first) 
     { 
      double[] doubleArray = (inst.decode(it.next().toString())); 
      lat = Double.toString(doubleArray[0]); 
      longi = Double.toString(doubleArray[1]); 
      first = false; 
     } 
    } 

你在接下来的while (it.hasNext())回路设置first = false;所以迭代if(first)没有输入和it.next()永远不会再次调用，所以如果it有不止一个元素it.hasNext()将总是返回true，并且您将永远不会离开这个while循环。