java - Hadoop - 查找 IP 命中总数和唯一 IP 地址，然后找到平均值(总 IP 命中数/uniqueIP)

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), 这样你就可以在 IPReducer 中定义 totalHits 和 uniqueIP 并且所有的 reduce() 调用都会看到这些变量的相同实例。然后，您可以在所有 reduce() 完成后运行的 reducer 的 cleanup() 方法中计算平均值。

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