高可用Hadoop平台-实战

  1. 云栖社区>
  2. 博客>
  3. 正文

高可用Hadoop平台-实战

smartloli 2016-04-19 17:01:28 浏览980
展开阅读全文

1.概述

  今天继续《高可用的Hadoop平台》系列,今天开始进行小规模的实战下,前面的准备工作完成后,基本用于统计数据的平台都拥有了,关于导出统计结果的文章留到后面赘述。今天要和大家分享的案例是一个基于电商网站的用户行为分析,这里分析的指标包含以下指标:

  • 统计每日PV
  • 每日注册用户
  • 每日IP
  • 跳出用户

  其他指标可以参考上述4个指标进行拓展,下面我们开始今天的分析之旅。

2.流程

  首先,在开发之前我们需要注意哪些问题?我们不能盲目的按照自己的意愿去开发项目,这样到头来得不到产品的认可,我们的工作又得返工。下面结合自身的工作,说说开发的具体流程:

  1. 需求产生过程及遇到问题和注意事项
  2. 产品组提出:报表需求+日志规范
  3. 报表研发流程(数据处理流程):网络日志(一般有专门的节点来存储日志)=>pull(拉取日志)=>本地存储(数据组可操作的服务器节点)=>预处理(清洗或过滤,存入到hive表)=>HDFS文件系统数据存储(统计的结果都存放在HDFS文件系统中)=>导出(将统计结构导出)=>Mysql数据库存储(或其他数据库,供前端人员展示)

  在日志拉取过程,所欲问题和注意事项:如果日志量不大,我们可以直接使用python脚本或shell脚本直接将日志上传到HDFS,若是海量数据,那我们可以使用 flume 进行上传。具体选择那种上传方式取决于实际的业务,可按需选择。

  注:若使用脚本上传,需考虑脚本的可读性和可维护性。

  在日志预处理过程中需要注意事项:对字段进行翻译,反编译,解析等操作,以确保存入到 hive 表的是有效的有用的信息。

  另外,在实际开发中,得和产品充分沟通过后,我们在开工;不然,到最后会引发一些不必要得麻烦。

  开发流程图和之前介绍《网站日志统计案例分析与实现》得流程图类似,这里直接拿过来使用,如下图所示:

3.开发

  开发流程我们清楚来,需要统计得指标任务也明确了,接下来,我们开始进行编码阶段。首先,这里我赘述得是安装流程图得过程来的,若是在实际开发过程中,可根据实际情况来定,可以先独立的开发后面的模块,预留接口功能。不作限制,按需开发。

3.1上传日志

  这里由于我本地只能连接到测试的集群上,而集群拉取的测试数据量很少,这里我就直接用 shell 脚本上传了。内容如下所示:


#! /bin/bash
# get date param yesterday=$1

hadoop dfs -put /hdfs/logs/day/$1 /hdfs/logs/day/

  然后上传脚本使用 crontab 来定时调度。

3.2清洗日志

  我们在确定 HDFS 存有数据后,对上传的日志进行清洗(或过滤),抽起对统计指标有用的数据源,并将数据源重定向到 HDFS 目录。,下面给出部分清洗代码,内容如下:

Map类:


/**
 * 
 */
package cn.hdfs.mapreducer;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import cn.jpush.hdfs.utils.LogParserFactory;

/**
 * @author dengjie
 * @date 2015年4月1日
 * @description TODO
 */
public class LogMapper extends Mapper<LongWritable, Text, LongWritable, Text> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    final String[] parsed = LogParserFactory.parse(value.toString());

    // 过滤掉静态信息
    if (parsed[2].startsWith("GET /static/") || parsed[2].startsWith("GET /uc_server")) {
        return;
    }

    // 过掉开头的特定格式字符串
    if (parsed[2].startsWith("GET /")) {
        parsed[2] = parsed[2].substring("GET /".length());
    } else if (parsed[2].startsWith("POST /")) {
        parsed[2] = parsed[2].substring("POST /".length());
    }

    // 过滤结尾的特定格式字符串
    if (parsed[2].endsWith(" HTTP/1.1")) {
        parsed[2] = parsed[2].substring(0, parsed[2].length() - " HTTP/1.1".length());
    }

    String str = "";
    for (int i = 0; i < parsed.length; i++) {
        if (i == (parsed.length - 1)) {
        str += parsed[i];
        } else {
        str += parsed[i] + ",";
        }
    }

    context.write(key, new Text(str));
    }

}

Reduce类:



/**
 * 
 */
package cn.hdfs.mapreducer;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * @author dengjie
 * @date 2015年4月1日
 * @description TODO
 */
public class LogReducer extends Reducer<LongWritable, Text, Text, NullWritable> {

    @Override
    protected void reduce(LongWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
    for (Text v : values) {
        context.write(v, NullWritable.get());
    }
    }

}

LogParserFactory类:



/**
 * 
 */
package cn.hdfs.utils;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;

/**
 * @author dengjie
 * @date 2015年4月1日
 * @description TODO
 */
public class LogParserFactory {
    public static final SimpleDateFormat FORMAT = new SimpleDateFormat("d/MMM/yyyy:HH:mm:ss", Locale.ENGLISH);
    public static final SimpleDateFormat dateformat = new SimpleDateFormat("yyyyMMddHHmmss");

    /**
     * 解析英文时间字符串
     * 
     * @param string
     * @return
     * @throws ParseException
     */
    private static Date parseDateFormat(String string) {
    Date parse = null;
    try {
        parse = FORMAT.parse(string);
    } catch (Exception e) {
        e.printStackTrace();
    }
    return parse;
    }

    /**
     * 解析日志的行记录
     * 
     * @param line
     * @return 数组含有5个元素,分别是ip、时间、url、状态、流量
     */
    public static String[] parse(String line) {
    String ip = parseIP(line);
    String time = parseTime(line);
    String url = parseURL(line);
    String status = parseStatus(line);
    String traffic = parseTraffic(line);

    return new String[] { ip, time, url, status, traffic };
    }

    private static String parseTraffic(String line) {
    final String trim = line.substring(line.lastIndexOf("\"") + 1).trim();
    String traffic = trim.split(" ")[1];
    return traffic;
    }

    private static String parseStatus(String line) {
    final String trim = line.substring(line.lastIndexOf("\"") + 1).trim();
    String status = trim.split(" ")[0];
    return status;
    }

    private static String parseURL(String line) {
    final int first = line.indexOf("\"");
    final int last = line.lastIndexOf("\"");
    String url = line.substring(first + 1, last);
    return url;
    }

    private static String parseTime(String line) {
    final int first = line.indexOf("[");
    final int last = line.indexOf("+0800]");
    String time = line.substring(first + 1, last).trim();
    Date date = parseDateFormat(time);
    return dateformat.format(date);
    }

    private static String parseIP(String line) {
    String ip = line.split("- -")[0].trim();
    return ip;
    }
}

Main函数:


/**
 * 
 */
package cn.hdfs.main;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import cn.jpush.hdfs.mapreducer.LogMapper;
import cn.jpush.hdfs.mapreducer.LogReducer;
import cn.jpush.hdfs.utils.ConfigUtils;

/**
 * @author dengjie
 * @date 2015年4月1日
 * @description 将清洗后的日志重新存放指定的hdfs上
 */
public class LogCleanMR extends Configured implements Tool {

    @SuppressWarnings("deprecation")
    public int run(String[] args) throws Exception {

    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", "hdfs://cluster1");
    conf.set("dfs.nameservices", "cluster1");
    conf.set("dfs.ha.namenodes.cluster1", "nna,nns");
    conf.set("dfs.namenode.rpc-address.cluster1.nna", "10.211.55.26:9000");
    conf.set("dfs.namenode.rpc-address.cluster1.nns", "10.211.55.27:9000");
    conf.set("dfs.client.failover.proxy.provider.cluster1", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");

    final Job job = new Job(conf, LogCleanMR.class.getSimpleName());
    job.setJarByClass(LogCleanMR.class);
    job.setMapperClass(LogMapper.class);
    job.setMapOutputKeyClass(LongWritable.class);
    job.setMapOutputValueClass(Text.class);
    job.setReducerClass(LogReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(NullWritable.class);
    FileInputFormat.setInputPaths(job, args[0]);
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    int status = job.waitForCompletion(true) ? 0 : 1;
    return status;
    }

    public static void main(String[] args) throws Exception {
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy_MM_dd");
    args = new String[] { ConfigUtils.HDFS.LOGDFS_PATH, String.format("/hdfs/logs/redirect/day/%s", sdf.format(new Date())) };
    int res = ToolRunner.run(new Configuration(), new LogCleanMR(), args);
    System.exit(res);
    }

}

  清洗工作到这里就已经完成了,下面开始统计指标任务的开发。

3.3统计指标

  在这里,由于 Java API 代码设计到实际的业务逻辑,我就直接使用 Hive SQL 来演示了统计结果了,若干有同学需要使用 Java API 来开发 Hive 应用,可参考《高可用Hadoop平台-集成Hive HAProxy》这篇博客,里面有讲到如何使用 Java API 来操作 Hive。下面我们使用 Hive SQL 来进行统计。内容如下:

建表:


CREATE EXTERNAL TABLE portal(ip string, atime string, url string,status int,traffic int)PARTITIONED BY (logdate string) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LOCATION '/hdfs/logdfs/portal'

创建分区:


ALTER TABLE portal ADD PARTITION(logdate='2015_01_02') 

加载数据:


LOCATION '/hdfs/logdfs/portal/2015_01_02'

 注:LOCATION 关键字后面的路径是指定清洗后的的hdfs路径

  下面创建临时统计表,各表如下所示:

创建每日PV表: 


CREATE TABLE pv_2015_01_02 AS SELECT COUNT(1) AS PV FROM logdfs WHERE logdate='2015_01_02';

创建注册用户表:


CREATE TABLE register_2015_01_02 AS SELECT COUNT(1) AS REGUSER FROM logdfs WHERE logdate='2015_01_02' AND INSTR(url,'signup')>0;

创建IP表:


CREATE TABLE ip_2015_01_02 AS SELECT COUNT(1) AS IP FROM (SELECT DISTINCT ip from logdfs WHERE logdate='2015_01_02') tmp

创建跳出用户表:


CREATE TABLE jumper_2015_01_02 AS SELECT COUNT(1) AS jumper FROM (SELECT COUNT(ip) AS times FROM logdfs WHERE logdate='2015_01_02' GROUP BY ip HAVING times=1) e;

最后我们将所有的结果汇总到一张 Hive 表,命令如下所示:


CREATE TABLE logdfs_2015_01_02 ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' AS SELECT '2015_01_02', a.pv, b.reguser, c.ip, d.jumper FROM pv_2015_01_02 a JOIN reguser_2015_01_02 b ON 1=1 JOIN ip_2015_01_02 c ON 1=1 JOIN jumper_2015_01_02 d ON 1=1 ;

 关于 JOIN ... ON 用法不熟悉的同学,可以参考《Hive基本操作》这篇文章。

4.总结

  这样,我们对使用 Hive 基于 HDFS 平台进行数据分析统计的流程就完成了,这里也许会发现一个问题,操作 Hive SQL 命令出错率是很高的,后面带我将业务从 Java API 分离出来后,我会将操作 Hive 的 Java API 贴在这篇博客的后面。至于如何将统计的结果导出,后面会花一篇博客来赘述导出的流程。

5.结束语

  这篇博客就和大家分享到这里,如果实际研究过程中有什么疑问,可以加群进行讨论或发送邮件给我,我会尽我所能为您解答,与君共勉!


网友评论

登录后评论
0/500
评论
smartloli
+ 关注