java - 使用 MapReduce 分析日志文件

Question

这是一个日志文件:

2011-10-26 06:11:35 user1 210.77.23.12
2011-10-26 06:11:45 user2 210.77.23.17
2011-10-26 06:11:46 user3 210.77.23.12
2011-10-26 06:11:47 user2 210.77.23.89
2011-10-26 06:11:48 user2 210.77.23.12
2011-10-26 06:11:52 user3 210.77.23.12
2011-10-26 06:11:53 user2 210.77.23.12
...

我想使用 MapReduce 按每行的第三个字段(用户)的记录次数降序排列。换句话说，我希望结果显示为:

user2 4
user3 2
user1 1

现在我有两个问题:

1. 默认情况下，MapReduce 将使用空格 和回车 分割日志文件，但是我只需要每一行的第三个字段，也就是说，我不关心2011-10-26、06:11:35、210.77等字段.23.12，如何让 MapReduce 忽略它们并提取 user 字段？

2. 默认情况下，MapReduce 将按键 而非值 对结果进行排序。如何让 MapReduce 按值(记录次数)对结果进行排序？

谢谢。

Answer 1

第一个问题:

您可能应该将整行传递给映射器，每次只保留第三个标记用于映射和映射 (user, 1)。

public class AnalyzeLogs
{       
    public static class FindFriendMapper extends Mapper<Object, Text, Text, IntWritable> {

    public void map(Object, Text value, Context context) throws IOException, InterruptedException 
    {       
        String tempStrings[] = value.toString().split(","); 
        context.write(new Text(tempStrings[2]), new IntWritable(1));
    }
}

对于你的第二个问题，我相信你无法避免在那之后有第二个 MR 工作(我想不出任何其他方式)。因此，第一个作业的 reducer 只会聚合值并为每个键给出一个总和，并按键排序。这还不是您所需要的。

因此，您将此作业的输出作为输入传递给第二个 MR 作业。这项工作的目的是在传递给 reducer 之前按值进行一些特殊的排序(这绝对不会做任何事情)。

我们的第二项工作的 Mapper 将如下所示:

public static class SortLogsMapper extends Mapper<Object, Text, Text, NullWritable> {

public void map(Object, Text value, Context context) throws IOException, InterruptedException 
{       
    context.write(value, new NullWritable());
}

如您所见，我们根本没有使用此映射器的值。相反，我们创建了一个包含我们的值的 key (我们的 key 采用key1 value1 格式)。现在剩下要做的是向框架指定它应该根据 value1 而不是整个 key1 value1 进行排序。所以我们将实现一个自定义的 SortComparator:

public static class LogDescComparator extends WritableComparator
{
    protected LogDescComparator() 
    {
        super(Text.class, true);
    }

    @Override
    public int compare(WritableComparable w1, WritableComparable w2)
    {

        Text t1 = (Text) w1;
        Text t2 = (Text) w2;
        String[] t1Items = t1.toString().split(" "); //probably it's a " "
        String[] t2Items = t2.toString().split(" ");
        String t1Value = t1Items[1];
        String t2Value = t2Items[1];
        int comp = t2Value.compareTo(t1Value); // We compare using "real" value part of our synthetic key in Descending order

        return comp;

    }
}

您可以将自定义比较器设置为:job.setSortComparatorClass(LogDescComparator.class);

工作的reducer应该什么都不做。但是，如果我们不设置 reducer，则映射器键的排序将不会完成(我们需要这样做)。因此，您需要将 IdentityReducer 设置为第二个 MR 作业的 Reducer 以不进行缩减，但仍确保映射器的合成键按照我们指定的方式排序。