使用Hadoop MapReduce分析邮件日志提取 id、状态 和 目标邮箱

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: 使用Hadoop MapReduce分析邮件日志提取 id、状态 和 目标邮箱

使用Hadoop MapReduce分析邮件日志提取 id、状态 和 目标邮箱

在大数据处理和分析的场景中,Hadoop MapReduce是一种常见且高效的工具。本文将展示如何使用Hadoop MapReduce来分析邮件日志,提取邮件的发送状态(成功、失败或退回)和目标邮箱。

项目结构

我们将创建一个Java项目,该项目包含三个主要部分:

**Mapper类:**解析邮件日志,提取ID、状态和目标邮箱。

**Reducer类:**汇总Mapper输出的数据,生成最终结果。

*Driver类:**配置和运行MapReduce作业。

数据格式

我们将处理的邮件日志示例如下:


在这些日志中,我们需要提取邮件的ID、发送状态(成功、失败或退回)和目标邮箱。

代码实现

以下是完整的Java代码,包含Mapper、Reducer和Driver类:

package org.example.mapReduce;

import java.io.IOException;
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.output.FileOutputFormat;

public class MailLogAnalysis {

    public static class MailLogMapper extends Mapper<LongWritable, Text, Text, Text> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();

            if (line.contains("starting delivery")) {
                String[] parts = line.split(" ");
                String id = parts[3].replace(":", "");
                String targetEmail = parts[8];
                context.write(new Text(id), new Text("email," + targetEmail));
            }

            if (line.contains("success") || line.contains("failure") || line.contains("bounce")) {
                String status = "success";
                if (line.contains("failure")) {
                    status = "failure";
                }
                if (line.contains("bounce")) {
                    status = "bounce";
                }
                String[] parts = line.split(" ");
                String id = parts[2].replace(":", "");
                context.write(new Text(id), new Text("status," + status));
            }
        }
    }

    public static class MailLogReducer extends Reducer<Text, Text, Text, Text> {
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            String email = "";
            String status = "failure";
            for (Text val : values) {
                String[] parts = val.toString().split(",", 2);
                if (parts[0].equals("email")) {
                    email = parts[1];
                } else if (parts[0].equals("status")) {
                    status = parts[1];
                }
            }
            context.write(key, new Text(status + "," + email));
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Mail Log Analysis");
        job.setJarByClass(MailLogAnalysis.class);
        job.setMapperClass(MailLogMapper.class);
        job.setReducerClass(MailLogReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

使用Hadoop MapReduce分析邮件日志

在大数据处理和分析的场景中,Hadoop MapReduce是一种常见且高效的工具。本文将展示如何使用Hadoop MapReduce来分析邮件日志,提取邮件的发送状态(成功、失败或退回)和目标邮箱。我们将通过一个具体的例子来实现这一目标。


项目结构

我们将创建一个Java项目,该项目包含三个主要部分:


Mapper类:解析邮件日志,提取ID、状态和目标邮箱。

Reducer类:汇总Mapper输出的数据,生成最终结果。

Driver类:配置和运行MapReduce作业。

数据格式

我们将处理的邮件日志示例如下:


less
复制代码
@400000004faa61e21e8e3e24 starting delivery 1820: msg 850901 to remote sunkang@189.cn
@400000004faa61e536864a44 delivery 1820: success: 121.14.53.136_accepted_message./Remote_host_said:_250_Ok:_queued_as_43A2222C006/
@400000004faa61e70a73c60c delivery 1823: deferral: 210.32.157.174_failed_after_I_sent_the_message./Remote_host_said:_450_Requested_action_not_taken:_AQAAf5CrT+qlYqpPamRUAA–.7571S2,_please_try_again/
@400000004faa61e70a73c60c bounce 1824: 550 Mailbox not found

在这些日志中,我们需要提取邮件的ID、发送状态(成功、失败或退回)和目标邮箱。


代码实现

以下是完整的Java代码,包含Mapper、Reducer和Driver类:


java
复制代码
package org.example.mapReduce;
import java.io.IOException;
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.output.FileOutputFormat;
public class MailLogAnalysis {
public static class MailLogMapper extends Mapper<LongWritable, Text, Text, Text> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();

        if (line.contains("starting delivery")) {
            String[] parts = line.split(" ");
            String id = parts[3].replace(":", "");
            String targetEmail = parts[8];
            context.write(new Text(id), new Text("email," + targetEmail));
        }

        if (line.contains("success") || line.contains("failure") || line.contains("bounce")) {
            String status = "success";
            if (line.contains("failure")) {
                status = "failure";
            }
            if (line.contains("bounce")) {
                status = "bounce";
            }
            String[] parts = line.split(" ");
            String id = parts[2].replace(":", "");
            context.write(new Text(id), new Text("status," + status));
        }
    }
}

public static class MailLogReducer extends Reducer<Text, Text, Text, Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        String email = "";
        String status = "failure";
        for (Text val : values) {
            String[] parts = val.toString().split(",", 2);
            if (parts[0].equals("email")) {
                email = parts[1];
            } else if (parts[0].equals("status")) {
                status = parts[1];
            }
        }
        context.write(key, new Text(status + "," + email));
    }
}

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "Mail Log Analysis");
    job.setJarByClass(MailLogAnalysis.class);
    job.setMapperClass(MailLogMapper.class);
    job.setReducerClass(MailLogReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);

    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));

    System.exit(job.waitForCompletion(true) ? 0 : 1);
}

代码解释

Mapper类

MailLogMapper类从日志中提取邮件的ID、目标邮箱和发送状态,并将这些信息作为键值对输出:


如果行包含"starting delivery",则提取邮件的ID和目标邮箱,并输出键值对<ID, email, 目标邮箱>。

如果行包含"success"、“failure"或"bounce”,则提取邮件的ID和发送状态,并输出键值对<ID, status, 发送状态>。

Reducer类

MailLogReducer类汇总Mapper输出的数据,生成最终的结果:


对于每个邮件ID,汇总对应的目标邮箱和发送状态。

输出包含ID、发送状态和目标邮箱的最终结果。

Driver类

MailLogAnalysis类配置和运行MapReduce作业:


设置作业名称、Mapper类和Reducer类。

设置输入路径和输出路径。

提交作业并等待完成。

MapReduce运行结果

总结

通过本文的示例,我们展示了如何使用Hadoop MapReduce来分析邮件日志,提取邮件的发送状态和目标邮箱。希望本文能为您的大数据处理和分析工作提供一些帮助。

相关实践学习
通过日志服务实现云资源OSS的安全审计
本实验介绍如何通过日志服务实现云资源OSS的安全审计。
相关文章
|
23天前
|
监控 安全 搜索推荐
使用EventLog Analyzer进行日志取证分析
EventLog Analyzer助力企业通过集中采集、归档与分析系统日志及syslog,快速构建“数字犯罪现场”,精准追溯安全事件根源。其强大搜索功能可秒级定位入侵时间、人员与路径,生成合规与取证报表,确保日志安全防篡改,大幅提升调查效率,为执法提供有力证据支持。
|
6月前
|
存储 运维 监控
SelectDB 实现日志高效存储与实时分析,完成任务可领取积分、餐具套装/水杯/帆布包!
SelectDB 实现日志高效存储与实时分析,完成任务可领取积分、餐具套装/水杯/帆布包!
|
6月前
|
SQL 监控 数据挖掘
SLS 重磅升级:超大规模数据实现完全精确分析
SLS 全新推出的「SQL 完全精确」模式,通过“限”与“换”的策略切换,在快速分析与精确计算之间实现平衡,满足用户对于超大数据规模分析结果精确的刚性需求。标志着其在超大规模日志数据分析领域再次迈出了重要的一步。
502 117
|
3月前
|
监控 安全 NoSQL
【DevOps】Logstash详解:高效日志管理与分析工具
Logstash是ELK Stack核心组件之一,具备强大的日志收集、处理与转发能力。它支持多种数据来源,提供灵活的过滤、转换机制,并可通过插件扩展功能,广泛应用于系统日志分析、性能优化及安全合规等领域,是现代日志管理的关键工具。
437 0
|
5月前
|
自然语言处理 监控 安全
阿里云发布可观测MCP!支持自然语言查询和分析多模态日志
阿里云可观测官方发布了Observable MCP Server,提供了一系列访问阿里云可观测各产品的工具能力,包含阿里云日志服务SLS、阿里云应用实时监控服务ARMS等,支持用户通过自然语言形式查询
546 0
阿里云发布可观测MCP!支持自然语言查询和分析多模态日志
|
4月前
|
人工智能 运维 监控
Aipy实战:分析apache2日志中的网站攻击痕迹
Apache2日志系统灵活且信息全面,但安全分析、实时分析和合规性审计存在较高技术门槛。为降低难度,可借助AI工具如aipy高效分析日志,快速发现攻击痕迹并提供反制措施。通过结合AI与学习技术知识,新手运维人员能更轻松掌握复杂日志分析任务,提升工作效率与技能水平。
|
7月前
|
存储 消息中间件 缓存
MiniMax GenAI 可观测性分析 :基于阿里云 SelectDB 构建 PB 级别日志系统
基于阿里云SelectDB,MiniMax构建了覆盖国内及海外业务的日志可观测中台,总体数据规模超过数PB,日均新增日志写入量达数百TB。系统在P95分位查询场景下的响应时间小于3秒,峰值时刻实现了超过10GB/s的读写吞吐。通过存算分离、高压缩比算法和单副本热缓存等技术手段,MiniMax在优化性能的同时显著降低了建设成本,计算资源用量降低40%,热数据存储用量降低50%,为未来业务的高速发展和技术演进奠定了坚实基础。
282 1
MiniMax GenAI 可观测性分析 :基于阿里云 SelectDB 构建 PB 级别日志系统
|
5月前
|
监控 容灾 算法
阿里云 SLS 多云日志接入最佳实践:链路、成本与高可用性优化
本文探讨了如何高效、经济且可靠地将海外应用与基础设施日志统一采集至阿里云日志服务(SLS),解决全球化业务扩展中的关键挑战。重点介绍了高性能日志采集Agent(iLogtail/LoongCollector)在海外场景的应用,推荐使用LoongCollector以获得更优的稳定性和网络容错能力。同时分析了多种网络接入方案,包括公网直连、全球加速优化、阿里云内网及专线/CEN/VPN接入等,并提供了成本优化策略和多目标发送配置指导,帮助企业构建稳定、低成本、高可用的全球日志系统。
635 54
|
10月前
|
监控 安全 Apache
什么是Apache日志?为什么Apache日志分析很重要?
Apache是全球广泛使用的Web服务器软件,支持超过30%的活跃网站。它通过接收和处理HTTP请求,与后端服务器通信,返回响应并记录日志,确保网页请求的快速准确处理。Apache日志分为访问日志和错误日志,对提升用户体验、保障安全及优化性能至关重要。EventLog Analyzer等工具可有效管理和分析这些日志,增强Web服务的安全性和可靠性。
265 9

热门文章

最新文章