减速器无法为不同的映射器按键分组

Question

用例：

1. 文件 1 包含展示数据，其中包含 trackerId + 其他字段
2. 文件 2 包含点击详情，包含 trackerId + clicked

我为以上两个和一个减速器使用不同的映射器，但减速器似乎无法合并两个文件数据。

package com.hadoop.intellipaat;

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

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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
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 com.google.common.collect.Lists;

/**
 * This job will combine click and impression on TrackerId
 * 
 * @author raghunandangupta
 *
 */

public class JoinClickImpressionDetailJob {

    public static final String IMPRESSION_PREFIX = "IMPRESSION_PREFIX";
    public static final String CLICK_PREFIX = "CLICK_PREFIX";
    public static final String SEPERATOR = "~";

    private static class ImpressionMapper extends Mapper<LongWritable, Text, Text, Text> {

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            /**
             * Excluding header
             */
            if (!(value.toString().indexOf("accountId") != -1)) {
                String words[] = value.toString().split(",");
                if (words.length > 18) {
                    context.write(new Text(words[18].trim()), new Text(IMPRESSION_PREFIX + SEPERATOR + value.toString()));
                }
            } else {
                context.write(new Text(""), value);
            }
        }
    }

    private static class ClickMapper extends Mapper<LongWritable, Text, Text, Text> {

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            String words[] = value.toString().split(",");
            if (words.length > 18) {
                context.write(new Text(words[18].trim()), new Text(CLICK_PREFIX + SEPERATOR + value.toString()));
            } else {
                context.write(new Text(""), new Text("1"));
            }
        }
    }

    private static class ImpressionClickReducer extends Reducer<Text, Text, Text, Text> {
        @Override
        protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) {
            try {
                System.out.println("=========="+key.toString());
                if (key.toString().length() != 0) {
                    List<Text> myList = Lists.newArrayList(values);

                    for(Text t : myList){
                        System.out.println("#######"+t.toString());
                    }
                    System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@");
                    if (myList.size() == 2) {
                        if (myList.get(0).toString().indexOf(IMPRESSION_PREFIX) != -1 && myList.get(1).toString().indexOf(CLICK_PREFIX) != -1) {
                            String line = myList.get(0).toString().split(SEPERATOR)[1] + ",1";
                            context.write(key, new Text(line));
                        } else if (myList.get(1).toString().indexOf(IMPRESSION_PREFIX) != -1
                                && myList.get(0).toString().indexOf(CLICK_PREFIX) != -1) {
                            String line = myList.get(1).toString().split(SEPERATOR)[1] + ",1";
                            context.write(key, new Text(line));
                        }
                    }
                }
            } catch (Exception exception) {
                exception.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        try {
            Configuration conf = new Configuration();
            // conf.set("mapreduce.output.fileoutputformat.compress", "true");
            // conf.set("mapreduce.output.fileoutputformat.compress.codec",
            // "org.apache.hadoop.io.compress.GzipCodec");
            // conf.set("mapreduce.map.output.compress.codec",
            // "org.apache.hadoop.io.compress.SnappyCodec");
            // conf.set("mapreduce.output.fileoutputformat.compress.type",
            // "BLOCK");
            Job job = Job.getInstance(conf, "IMPRESSION_CLICK_COMBINE_JOB");

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);

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

            job.setReducerClass(ImpressionClickReducer.class);

            FileInputFormat.setInputDirRecursive(job, true);

            // FileInputFormat.addInputPath(job, new Path(args[0]));
            // job.setMapperClass(ImpressionMapper.class);

            /**
             * Here directory of impressions will be present
             */
            MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, ImpressionMapper.class);
            /**
             * Here directory of clicks will be present
             */
            MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, ClickMapper.class);

            FileOutputFormat.setOutputPath(job, new Path(args[2]));

            job.waitForCompletion(true);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

任何线索将不胜感激。

例如。 File 1 [trackerId1,record1] File2 [treackerId1, Clicked]

在减速器中我得到：

trackerId,[record1,record1] 理想情况下应该是trackerId ,[record1,clicked]

Answer 1

您的问题很可能与减速器中的这一行有关：

List<Text> myList = Lists.newArrayList(values);

要记住的主要事情是Iterable<Text> values 正在重用它在您迭代时提供给您的Text 对象。因此，您可能会在数组中添加两个 Text 对象，但它们指向的是同一个对象。

如果您查看Lists.newArrayList() 的工作原理，它只是将对象添加到数组中，而不创建新对象。

因此，如果您要使用Text 对象，则每次向数组添加值时都需要创建一个新对象。这通常是人们在这种情况下使用字符串的原因。快速检查这是否是问题所在，可以将此代码更改为：

List<Text> myList = new ArrayList<Text>();
for (Text v : values) {
    myList.add(new Text(v));
}

因此，您每次都创建一个新的Text。