Hadoop - 查找 IP 命中总数和唯一 IP 地址，然后找到平均值（总 IP 命中/唯一 IP）

Question

我正在努力学习 Hadoop。我编写了一个 map reduce 代码，用于查找 IP 命中总数并查找唯一 IP 地址，然后找到平均值（总 IP 命中/唯一 ID）。

但是，我得到了所有 IP 的输出以及点击次数。但我无法得到相同的平均值。

代码：

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public final class IPAddress {
    private final static IntWritable ONE = new IntWritable(1);

    static int totalHits = 0, uniqueIP = 0;
    public final static void main(final String[] args) throws Exception 
    {
        final Configuration conf = new Configuration();

        final Job job = new Job(conf, "IPAddress");
        job.setJarByClass(IPAddress.class);

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

        job.setMapperClass(IPMap.class);
        job.setCombinerClass(IPReduce.class);
        job.setReducerClass(IPReduce.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);
        int average = totalHits/uniqueIP;
        System.out.print("Average is :"+average+"\n");
    }

    public static final class IPMap extends Mapper<LongWritable, Text, Text, IntWritable> 
    {
        private final Text mapKey = new Text();

        public final void map(final LongWritable key, final Text value, final Context context) throws IOException, InterruptedException 
        {
            final String line = value.toString();
            final String[] data = line.trim().split("- -");
            if (data.length > 1) 
            {
                final String ipAddress = data[0];
                mapKey.set(ipAddress);
                context.write(mapKey, ONE);
            }
        }
    }

    public static final class IPReduce extends Reducer<Text, IntWritable, Text, IntWritable> 
    {

        public final void reduce(final Text key, final Iterable<IntWritable> values, final Context context) throws IOException, InterruptedException 
        {
            int count = 0, sum = 0, distinctIpCount=0;
            for (final IntWritable val : values) 
            {
                count += val.get();
                sum += count;
                distinctIpCount++;
            }
            totalHits = count;
            uniqueIP = distinctIpCount;
            context.write(key, new IntWritable(count));


        }
    }
}

Answer 1

关于 MapReduce 作业执行的重要一点是，即使您在一个类中提供所有代码，MapReduce 框架也会提取您提供的映射器和归约器类并将它们发送到工作节点执行，而main() 方法在您开始工作的本地 JVM 上运行。这意味着 mapper 和 reducer 类方法无法看到您在 mapper 和 reducer 类之外定义的任何变量。

具体到您的用例，如果您想计算所有 IP 地址的平均命中，您可以在调用作业时仅使用一个 reducer (-D mapred.reduce.tasks=1)，这样您就可以定义 totalHits 和 @ IPReducer 中的 987654325@ 和所有 reduce() 调用将看到这些变量的相同实例。然后，您可以在所有 reduce()s 都完成后运行的 reducer 的 cleanup() 方法中计算您的平均值。

您将无法轻松地将其发送回主程序以打印到屏幕上，但您可以将结果输出为作业输出（提供相同的 Context 对象），或者如果您要将每个 IP 计数保留为主要作业输出，请使用 HDFS API 将平均值写入 HDFS 文件。