Hadoop MapReduce 统计日志文件中的 IP 访问次数

需求分析
日志文件中每一行通常包含客户端 IP 地址（如 192.168.1.1 - - [10/Oct/2023:13:55:36] "GET /index.html HTTP/1.1" 200 2326）。目标是通过 MapReduce 统计每个 IP 出现的次数。

---

实现步骤

Mapper 阶段
从日志行中提取 IP 地址，将其作为 key 输出，value 设为计数 1（格式：<IP, 1>）。
示例代码（Java）：

public class LogMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text ip = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) 
            throws IOException, InterruptedException {
        String line = value.toString();
        // 假设 IP 是每行第一个字段（以空格分隔）
        String[] parts = line.split(" ");
        if (parts.length > 0) {
            ip.set(parts[0]);
            context.write(ip, one);
        }
    }
}

Reducer 阶段
对相同 IP 的计数进行累加，输出 <IP, total_count>。
示例代码：

public class LogReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) 
            throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        context.write(key, new IntWritable(sum));
    }
}

Driver 主类
配置并提交 MapReduce 任务：

public class LogDriver extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(getConf(), "IP Count");
        job.setJarByClass(LogDriver.class);
        
        job.setMapperClass(LogMapper.class);
        job.setReducerClass(LogReducer.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new LogDriver(), args);
        System.exit(exitCode);
    }
}

---

日志格式适配建议

若日志格式复杂（如包含时间戳、URL 等），需调整 Mapper 的解析逻辑：

• 使用正则表达式精准提取 IP（如 ^(\d{1,3}\.){3}\d{1,3}）。
• 处理异常行（如跳过不符合格式的行）。

---

执行与输出

1. 打包 JAR 文件并上传到 Hadoop 集群。
2. 运行命令：

   hadoop jar ipcount.jar LogDriver /input/logs /output/ip_count
   

3. 结果文件将包含类似内容：

   192.168.1.1    15
   192.168.1.2    8
   

---

优化方向

• Combiner 阶段：在 Mapper 后本地聚合，减少数据传输。

  job.setCombinerClass(LogReducer.class);
  

• 分区优化：自定义 Partitioner 均匀分配 IP 到 Reducer。
• 输入格式：针对大文件使用 CombineTextInputFormat 合并小文件。