高可用Hadoop平台－实战

2016-04-18 445

1.概述

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

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

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

2.流程

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

需求产生过程及遇到问题和注意事项
产品组提出：报表需求＋日志规范
报表研发流程（数据处理流程）：网络日志（一般有专门的节点来存储日志）=>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.结束语

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

微信关注我们

原文链接：https://yq.aliyun.com/articles/31012

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

高可用Hadoop平台－实战尾声篇

1.概述今天这篇博客就是《高可用Hadoop平台》的尾声篇了，从搭建安装到入门运行 Hadoop 版的 HelloWorld（WordCount 可以称的上是 Hadoop 版的 HelloWorld ），在到开发中需要用到的各个套件以及对套件的安装使用，在到 Hadoop 的实战，一路走来我们对在Hadoop平台下开发的基本流程应该都熟悉了。今天我们来完成在高可用Hadoop平台开发的最后一步，导出数据。 2.导出数据目的首先，我来说明下为什么要导出数据，导出数据的目的是为了干嘛？我们都知道，我们当初统计这些数据的目标，就是为了来可视化这些数据结果；虽然结果我们是统计出来储存在 HDFS 上，但是，前段同学需要拿到这些数据，直接操作 HDFS 读取统计结果，这样的做法是不明智的，撇开安全性不说，时延就是一个很大的问题。所以，这里我们需要有一个步骤去完成数据的导出，将数据导出到 Mysql 之类的关系型数据库。这里我们用到的导出套件是Sqoop。 3.Sqoop 3.1安装包 sqoop下载地址 3.2配置打开配置环境文件： sudo vi /etc/profile 编辑 Sq...

2016-04-19

535

InputFormat描述了一个Map-Reduce作业中的输入规范。Map-Reduce框架依靠作业的InputFormat实现以下内容： 1、校验作业的输入规范； 2、分割输入文件（可能为多个），生成逻辑输入分片InputSplit（往往为多个），每个输入分片InputSplit接着被分配给单独的Mapper； 3、提供记录读取器RecordReader的实现，RecordReader被用于从逻辑输入分片InputSplit收集输入记录，这些输入记录会被交由Mapper处理。基于文件的输入格式的默认行为，作为代表性的子类FileInputFormat，基于输入文件的总大小（单位byte）来切分成逻辑输入分片InputSplit。然而，输入文件的文件系统数据块大小，被用作输入分片大小的上界。输入分片大小的下界则可以在mapred-default.xml配置文件中通过参数mapreduce.input.fileinputformat.split.minsize来配置。无疑，由于记录界限应该被遵守，基于输入大小的逻辑输入分片不满足很多应用。在这种情...

2016-04-19

633

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。