Mapreduce任务实现邮件监控

Mapreduce任务实现邮件监控

    这里主要使用Java自带邮件类实现Mapreduce任务的监控,如果Mapreduce任务报错则发送报错邮件。Mapreduce的报错信息通过hdfs中的日志获取,里面的报错日志是json格式,这里先将json转换成xml格式然后再发送到邮件。具体代码如下

import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.net.URI;
import java.util.Properties;
import java.util.StringTokenizer;

import javax.mail.Authenticator;
import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.PasswordAuthentication;
import javax.mail.Session;
import javax.mail.Transport;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;

import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import net.sf.json.xml.XMLSerializer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;

public class Email {

    private static final String USERNAME = "123456@qq.com";//发送邮件的用户名
    private static final String PASSWORD = "123456789";//发送邮件的用户名对应的密码
    private static final String EMAIL_HOST = "smtp.qq.com";//邮件服务器host

    public static void main(String args[]) {
        try {
            sendEmail("测试邮件", "测试邮件内容!", "test@qq.com");
            System.out.println("email ok !");
        } catch (MessagingException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    /**
     * @category 发送邮件方法,该方法实现发送Mapreduce任务报错信息,具体的报错信息通过hdfs的报错日志获取
     * @param to 目标邮箱(可以多个邮箱,用,号隔开)
     * @param job 通过mapreduce的job获取jobID
     * @param time 通过时间戳访问错误日志路径
     * @throws Exception
     */
    public static void sendErrMail(String to, Job job, String time)
            throws Exception {
        String subject = job.getJobName();
        String message = getErr(job, time);
        LoginMail lm = new LoginMail(USERNAME, PASSWORD);
        // 创建session
        Properties props = new Properties();
        props.put("mail.smtp.auth", "true");
        props.put("mail.smtp.host", EMAIL_HOST);
        Session session = Session.getDefaultInstance(props, lm);

        // 创建 message
        Message msg = new MimeMessage(session);

        // 设置发送源地址
        msg.setFrom(new InternetAddress(USERNAME));

        // 多用户分解
        StringTokenizer st = new StringTokenizer(to, ",");
        String[] recipients = new String[st.countTokens()];
        int rc = 0;
        while (st.hasMoreTokens())
            recipients[rc++] = st.nextToken();
        InternetAddress[] addressTo = new InternetAddress[recipients.length];
        for (int i = 0; i < recipients.length; i++) {
            addressTo[i] = new InternetAddress(recipients[i]);
        }
        msg.setRecipients(Message.RecipientType.TO, addressTo);

        // 设置邮件主题并发送邮件
        msg.setSubject(subject);
        msg.setContent(message, "text/html;charset=utf-8");
        Transport.send(msg);
    }

    /**
     * @category 自定义主题内容发送,这里的邮件内容不一定是Mapreduce的,可以任意填写
     * @param subject 主题
     * @param body 内容
     * @param to 目标邮箱
     * @throws MessagingException
     */
    public static void sendEmail(String subject, String body, String to)
            throws MessagingException {
        LoginMail lm = new LoginMail(USERNAME, PASSWORD);
        // 创建session
        Properties props = new Properties();
        props.put("mail.smtp.auth", "true");
        props.put("mail.smtp.host", EMAIL_HOST);
        Session session = Session.getDefaultInstance(props, lm);

        // 创建 message
        Message msg = new MimeMessage(session);

        // 设置发送源地址
        msg.setFrom(new InternetAddress(USERNAME));

        // 多用户分解
        StringTokenizer st = new StringTokenizer(to, ",");
        String[] recipients = new String[st.countTokens()];
        int rc = 0;
        while (st.hasMoreTokens())
            recipients[rc++] = st.nextToken();
        InternetAddress[] addressTo = new InternetAddress[recipients.length];
        for (int i = 0; i < recipients.length; i++) {
            addressTo[i] = new InternetAddress(recipients[i]);
        }
        msg.setRecipients(Message.RecipientType.TO, addressTo);

        // 设置邮件主题并发送邮件
        msg.setSubject(subject);
        msg.setContent(body, "text/html;charset=utf-8");
        Transport.send(msg);

    }

    /**
     * @category 获取日志文件
     * @param job
     * @param time
     * @return FSDataInputStream
     * @throws IOException
     */
    public static FSDataInputStream getFile(Job job, String time)
            throws IOException {
        String year = time.substring(0, 4);
        String month = time.substring(4, 6);
        String day = time.substring(6, 8);
        String dst = "hdfs://192.168.1.100:9000/tmp/hadoop-yarn/staging/history/done/"
                + year + "/" + month + "/" + day + "/000000";
        FileSystem fs = FileSystem.get(URI.create(dst), new Configuration());
        FileStatus[] status = fs.listStatus(new Path(dst));
        FSDataInputStream in = null;
        for (int i = 0; i < status.length; i++) {
            if (status[i].getPath().getName()
                    .contains(job.getJobID().toString())
                    && status[i].getPath().getName().endsWith("jhist")) {
                in = new FSDataInputStream(fs.open(status[i].getPath()));
            }
        }
        return in;
    }

    /**
     * @category 解析文件类容为xml
     * @param job
     * @param time
     * @return xml
     * @throws IOException
     * @throws InterruptedException
     */
    public static String getErr(Job job, String time) throws IOException,
            InterruptedException {
        FSDataInputStream in = getFile(job, time);
        Thread t1 = new Thread();
        while (in == null) {
            t1.sleep(20000);//由于hdfs每个job的日志不是实时生成,所以需要每隔20秒检查一次hdfs该job日志是否已生成
            t1.join();
            in = getFile(job, time);
        }
        BufferedReader br = new BufferedReader(new InputStreamReader(in));

        String line = "";
        JSONObject jo;
        JSONArray jsa = new JSONArray();
        String xml = "";
        XMLSerializer xmlSerializer = new XMLSerializer();
        while ((line = br.readLine()) != null) {
            if (line.toUpperCase().indexOf("error".toUpperCase()) > -1) {
                jo = JSONObject.fromObject(line);
                jsa.add(jo);
            }
        }
        xml = xmlSerializer.write(jsa);
        in.close();
        br.close();
        return xml;

    }

    /**
     * @category 获取try-catch中的异常内容
     * @param e Exception
     * @return 异常内容
     */
    public static String getException(Exception e) {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        PrintStream pout = new PrintStream(out);
        e.printStackTrace(pout);
        String ret = new String(out.toByteArray());
        pout.close();
        try {
            out.close();
        } catch (Exception ex) {
        }
        return ret;
    }
}

class LoginMail extends Authenticator {

    private String username;
    private String password;

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    @Override
    protected PasswordAuthentication getPasswordAuthentication() {
        return new PasswordAuthentication(username, password);
    }

    public LoginMail(String username, String password) {
        this.username = username;
        this.password = password;
    }
}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏数据结构与算法

筛法求素数 6分

11:回文素数 查看 提交 统计 提问 总时间限制: 5000ms 内存限制: 65536kB描述一个数如果从左往右读和从右往左读数字是相同的,则称这个数是回文...

35410
来自专栏编码小白

tomcat源码解读四 tomcat中的processer

     Processor是一个接口,针对于不同协议下具有不同的具体实现类,其实现类的具体功能是处理http请求,主要是对协议进行解析,状态处理以及响应。然后...

4567
来自专栏编码小白

tomcat请求处理分析(六)servlet的处理过程

1.1.1.1  servlet的解析过程 servlet的解析分为两步实现,第一个是匹配到对应的Wrapper,第二个是加载对应的servlet并进行数据,这...

7107
来自专栏Hongten

spring+hibernate+JQuery开发_电子相册_源码

=============================================================

4404
来自专栏码匠的流水账

频繁产生对象造成gc时间过长案例分析

gc时间过长,平均gc pause的时间要将近4秒,有13%的gc超过10秒,太可怕了,部分gc日志如下:

1901
来自专栏java一日一条

50个常见的 Java 错误及避免方法(第三部分)

当我们尝试调用带有错误参数的Java代码时,通常会产生此Java错误消息(@ghacksnews):

1483
来自专栏魂祭心

原 结合源码分析 setTimeout /

3656
来自专栏菩提树下的杨过

silverlight动态读取txt文件/解析json数据/调用wcf示例

终于开始正式学习silverlight,虽然有点晚,但总算开始了,今天看了一下sdk,主要是想看下silverlight中如何动态调用数据,对于数据库的访问,s...

23210
来自专栏Java成神之路

Java企业微信开发_05_消息推送之被动回复消息

微信加解密包 下载地址:http://qydev.weixin.qq.com/java.zip      ,此包中封装好了AES加解密方法,直接调用方法即可。

3082
来自专栏码匠的流水账

JCTools简介

JCTools是一款对jdk并发数据结构进行增强的并发工具,主要提供了map以及queue的增强数据结构。原来netty还是自己写的MpscLinkedQueu...

1661

扫码关注云+社区

领取腾讯云代金券