1

我不断增加减速器的数量，我发现虽然除了一个减速器之外的所有减速器都快速运行并完成它们的工作，但最后一个减速器只是挂在合并步骤，并在其 tasktracker 日志中显示以下消息：

Down to the last merge-pass, with 3 segments left of total size: 171207264 bytes

...并且在这个语句停留了很长时间之后，它抛出了一个 java 堆错误并开始了一些没有完成的清理。

我将 child.opts 内存增加到 3.5GB（无法超出此限制）并压缩地图输出。

可能是什么原因？

下面是驱动代码：

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    conf.set("mapred.task.timeout", "6000000");
    conf.set("mapred.compress.map.output", "true");
    Job job = new Job(conf, "FreebasePreprocess_Phase2");
    job.setNumReduceTasks(6);
    job.setJarByClass(FreebasePreprocess.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Text.class);

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

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

    FileInputFormat.addInputPath(job, new Path("/user/watsonuser/freebase_data100m120m_output"));
    FileOutputFormat.setOutputPath(job, new Path("/user/watsonuser/freebase_data100m120m_output_2"));

    job.waitForCompletion(true);
}

这是映射器：

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;


public class Map extends Mapper<LongWritable, Text, Text, Text>{

public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
{
    String[] entities = value.toString().split("\\t");
    String[] strings = {"/type/object/type", "/common/topic/notable_for", "/type/user/usergroup"};
    List<String> filteredPredicates = Arrays.asList(strings);

    FileSplit fileSplit = (FileSplit)context.getInputSplit();
    String filename = fileSplit.getPath().getName();
    //      System.out.println("File name "+filename);

    if(filename.startsWith("part-r")) {
        //      if(filename.equalsIgnoreCase("quad.tsv")) {
        //this is a quad dump file
        String name = null;
        String predicate = null;
        String oid = null;
        String outVal = null;
        String outKey = null;
        if(entities.length==3) {
            oid = entities[0].trim();
            predicate = entities[1].trim();
            name = entities[2].trim();

            /*if(predicate.contains("/type/object/name/lang"))
            {
                if(predicate.endsWith("/en")) 
                {*/
                /*outKey = sid;
                outVal = oid+"#-#-#-#"+"topic_name";
                context.write(new Text(outKey), new Text(outVal));*/
            /*  }
            }*/
                outKey = oid;
                outVal = predicate+"#-#-#-#"+name;
                context.write(new Text(outKey), new Text(outVal));

        }
    }

    else if(filename.equalsIgnoreCase("freebase-simple-topic-dump.tsv")) {
        //this is a simple topic dump file
        String sid = null;
        String name = null;
        String outKey = null;
        String outVal = null;
        if(entities.length>1) {
            sid = entities[0];
            name = entities[1];
            outKey = sid;
            outVal = name+"#-#-#-#"+"topic_name";
            context.write(new Text(outKey), new Text(outVal));
        }
    }
}

}

这是减速机

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;


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

public void reduce(Text key, Iterable<Text> values, Context context) 
        throws IOException, InterruptedException 
        {
            String name = null; 
            String sid = null;
            String predicate = null;
            String oid = null;
            String id = null;
            String outKey = null;
            String outVal = null;

            ArrayList<Text> valuesList = new ArrayList<Text>();
            Iterator<Text> ite = values.iterator();
            while(ite.hasNext()) {
                Text t = ite.next();
                Text txt = new Text();
                txt.set(t.toString());
                valuesList.add(txt);
                String[] entities = t.toString().split("#-#-#-#");
                if(entities[entities.length-1].equalsIgnoreCase("topic_name"))
                {
                    name = entities[0];
                }
            }

            for(int i=0; i<valuesList.size(); i++) { 
{ 

                Text t2 = valuesList.get(i);
                String[] entities = t2.toString().split("#-#-#-#");
                if(!entities[entities.length-1].contains("topic_name"))
                {
                    if(name!=null) {
                        outKey = entities[1]+"\t"+entities[0]+"\t"+name;
                    }
                    else {
                        outKey = entities[1]+"\t"+entities[0]+"\t"+key.toString();
                    }
                    context.write(new Text(outKey), null);
                }
            }
        }
}
4

2 回答 2

1

我的猜测是，您有一个具有大量值的键，并且减速器中的以下行会导致您出现问题：

valuesList.add(txt);

假设您有一个具有 100m 值的键，您正在尝试构建一个大小为 100m 的数组列表 - 在某个阶段您的减速器 JVM 将耗尽内存。

您可以通过进行一些调试并检查永不结束的减速器的日志来确认这一点：

valuesList.add(txt);
if (valuesList.size() % 10000 == 0) {
  System.err.println(key + "\t" + valueList.size());
}
于 2013-03-21T21:24:28.030 回答
0

我已经有一段时间没有写原始 MR 了，但我会以类似于以下的方式来处理它：

将键的所有值保存在内存中总是很危险的。相反，我会在您的工作中添加另一个 MR 阶段。在第一阶段，当 value 包含“topic-name”时发出 newkey = (key, 0), newValue = value，当 value 不包含“topic-name”时发出 newkey = (key, 1), newValue = value。这将需要编写一个可以处理一对的自定义 writablecomparable，并且知道如何对其进行排序。

对于下一阶段的化简器，编写一个分区器，在新键的第一个元素上进行分区。现在由于最后一个 reducer 的按键排序输出，您可以保证在获得每个键的其他 k,v 对之前获得带有 'name' 的 k,v 对。现在您可以访问与键对应的每个值的“名称”。

于 2013-03-24T23:12:02.363 回答