专栏首页Albert陈凯Hadoop数据分析平台实战——260用户数据ETL离线数据分析平台实战——260用户数据ETL

Hadoop数据分析平台实战——260用户数据ETL离线数据分析平台实战——260用户数据ETL

离线数据分析平台实战——260用户数据ETL

ETL目标

解析我们收集的日志数据,将解析后的数据保存到hbase中。 这里选择hbase来存储数据的主要原因就是: hbase的宽表结构设计适合我们的这样多种数据格式的数据存储(不同event有不同的存储格式)。 在etl过程中,我们需要将我们收集得到的数据进行处理,包括ip地址解析、userAgent解析、服务器时间解析等。

ETL存储

etl的结果存储到hbase中, 由于考虑到不同事件有不同的数据格式, 所以我们将最终etl的结果保存到hbase中, 我们使用单family的数据格式, rowkey的生产模式我们采用timestamp+uuid.crc编码的方式。 hbase创建命令:create 'event_logs', 'info'

操作步骤

  1. 修改pom文件,添加hadoop和hbase依赖
  2. 添加LoggerUtil类,中间设计到EventLogConstant常量类和TimeUtil工具类 LoggerUtil主要作用就是解析日志,返回一个map对象 EventLogConstants主要作用就是描述hbase的event_logs表的信息(表名,列簇名,列名)以及日志收集的日志中的数据参数name。其实列名和name是一样的。 TimeUtil主要作用解析服务器时间以及定义rowkey中的timestamp时间戳格式。
  3. 编写mapper类和runner类
  4. 添加环境变量文件,core-site.xml hbase-site.xml log4j.properties 根据不同的运行情况,修改源码将修改后的源码放到代码中。
  5. 添加pom编译代码,并进行测试 本地运行测试: 需要注意的就是windows环境可能会导致出现access方法调用异常,需要修改nativeio这个java文件。 使用TableMapReduceUtil的时候如果出现异常:*****/htrace-core-2.04.jar from hdfs://***/htrace-core-2.04.jar is not a valid DFS filename. 就需要将addDependencyJars参数设置为false。
本地提交job,集群运行测试:

    本地需要知道提交的job是需要提交到集群上的,所以需要指定两个参数mapreduce.framework.name和yarn.resourcemanager.address,value分别为yarn和hh:8032即可,但是可能会出现异常信息,此时需要将参数mapreduce.app-submission.cross-platform设置为true。

参数设置:

        mapreduce.framework.name=yarn
        yarn.resourcemanager.address=hh:8032
        mapreduce.app-submission.cross-platform=true

异常:

        1. Permission denied: user=gerry, access=EXECUTE, inode="/tmp":hadoop:supergroup:drwx------
            解决方案:执行hdfs dfs -chmod -R 777 /tmp
        2. Stack trace: ExitCodeException exitCode=1: /bin/bash: line 0: fg: no job control
            解决方案:添加mapreduce.app-submission.cross-platform=true
        3. ExitCodeException exitCode=1:
            解决方案:habse指定输出reducer的时候必须给定addDependencyJars参数为true。
        4. Class com.beifeng.etl.mr.ald.AnalyserLogDataMapper not found
            解决方案:引入EJob.java文件,然后再runner类中添加代码
                File jarFile = EJob.createTempJar("target/classes");
                ((JobConf) job.getConfiguration()).setJar(jarFile.toString());

集群提交&运行job测试:

package com.bjsxt.ae.util;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;

/**
 * 时间控制工具类
 * 
 * @author ck
 *
 */
public class TimeUtil {
    
    
    public static final String DATE_FORMAT = "yyyy-MM-dd";

    /**
     * 获取昨日的日期格式字符串数据
     * 
     * @return
     */
    public static String getYesterday() {
        return getYesterday(DATE_FORMAT);
    }

    /**
     * 获取对应格式的时间字符串
     * 
     * @param pattern
     * @return
     */
    public static String getYesterday(String pattern) {
        SimpleDateFormat sdf = new SimpleDateFormat(pattern);
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.DAY_OF_YEAR, -1);
        return sdf.format(calendar.getTime());
    }

    /**
     * 判断输入的参数是否是一个有效的时间格式数据
     * 
     * @param input
     * @return
     */
    public static boolean isValidateRunningDate(String input) {
        Matcher matcher = null;
        boolean result = false;
        String regex = "[0-9]{4}-[0-9]{2}-[0-9]{2}";
        if (input != null && !input.isEmpty()) {
            Pattern pattern = Pattern.compile(regex);
            matcher = pattern.matcher(input);
        }
        if (matcher != null) {
            result = matcher.matches();
        }
        return result;
    }

    /**
     * 将yyyy-MM-dd格式的时间字符串转换为时间戳
     * 
     * @param input
     * @return
     */
    public static long parseString2Long(String input) {
        return parseString2Long(input, DATE_FORMAT);
    }

    /**
     * 将指定格式的时间字符串转换为时间戳
     * 
     * @param input
     * @param pattern
     * @return
     */
    public static long parseString2Long(String input, String pattern) {
        Date date = null;
        try {
            date = new SimpleDateFormat(pattern).parse(input);
        } catch (ParseException e) {
            throw new RuntimeException(e);
        }
        return date.getTime();
    }

    /**
     * 将时间戳转换为yyyy-MM-dd格式的时间字符串
     * @param input
     * @return
     */
    public static String parseLong2String(long input) {
        return parseLong2String(input, DATE_FORMAT);
    }

    /**
     * 将时间戳转换为指定格式的字符串
     * 
     * @param input
     * @param pattern
     * @return
     */
    public static String parseLong2String(long input, String pattern) {
        Calendar calendar = Calendar.getInstance();
        calendar.setTimeInMillis(input);
        return new SimpleDateFormat(pattern).format(calendar.getTime());
    }

    /**
     * 将nginx服务器时间转换为时间戳,如果说解析失败,返回-1
     * 
     * @param input
     * @return
     */
    public static long parseNginxServerTime2Long(String input) {
        Date date = parseNginxServerTime2Date(input);
        return date == null ? -1L : date.getTime();
    }

    /**
     * 将nginx服务器时间转换为date对象。如果解析失败,返回null
     * 
     * @param input
     *            格式: 1449410796.976
     * @return
     */
    public static Date parseNginxServerTime2Date(String input) {
        if (StringUtils.isNotBlank(input)) {
            try {
                long timestamp = Double.valueOf(Double.valueOf(input.trim()) * 1000).longValue();
                Calendar calendar = Calendar.getInstance();
                calendar.setTimeInMillis(timestamp);
                return calendar.getTime();
            } catch (Exception e) {
                // nothing
            }
        }
        return null;
    }
}

package com.bjsxt.ae.util.model;

/**
 * 地区的pojo
 * @author root
 *
 */
public class RegionInfo {
    
    public static final String DEFAULT_VALUE = "unknown"; //默认值
    
    //国家
    private String country = DEFAULT_VALUE;
    //省
    private String province = DEFAULT_VALUE;
    //城市
    private String city = DEFAULT_VALUE;

    public String getCountry() {
        return country;
    }

    public void setCountry(String country) {
        this.country = country;
    }

    public String getProvince() {
        return province;
    }

    public void setProvince(String province) {
        this.province = province;
    }

    public String getCity() {
        return city;
    }

    public void setCity(String city) {
        this.city = city;
    }

    @Override
    public String toString() {
        return "RegionInfo [country=" + country + ", province=" + province + ", city=" + city + "]";
    }
            

}

package com.bjsxt.ae.etl.util;

import com.bjsxt.ae.etl.util.ip.IPSeeker;

/**
 * 定义具体的ip解析的类,最终调用IpSeeker类(父类)<br/>
 * 解析ip最终的返回时:国家名称 省份名称 城市名称<br/>
 * 如果是国外的ip,那么直接设置为unknown unknown unknown<br/>
 * 如果是国内ip,如果没法进行解析,那么就设置为中国 unknown unknown<br/>
 * 
 * @author ck
 *
 */
public class IPSeekerExt extends IPSeeker {
    private RegionInfo DEFAULT_INFO = new RegionInfo();
    
    
    public static void main(String[] args) {
        
        String ip = new String ("114.114.114.114");
        
        IPSeekerExt ips = new IPSeekerExt();
        System.out.println(ips.getCountry(ip));
        
        
    }

    /**
     * 解析ip地址,返回该ip地址对应的国家省份信息<br/>
     * 如果该ip解析失败,那么直接返回默认值
     * 
     * @param ip
     *            要解析的ip地址,格式为: 120.197.87.216
     * @return
     */
    public RegionInfo analyticIp(String ip) {
        if (ip == null || ip.trim().isEmpty()) {
            return DEFAULT_INFO;
        }

        RegionInfo info = new RegionInfo();
        try {
            String country = super.getCountry(ip);
            if (country != null && !country.trim().isEmpty()) {
                // 表示该ip还一个可以解析的ip
                country = country.trim();
                int length = country.length();
                int index = country.indexOf('省');
                if (index > 0) {
                    // 当前ip属于23个省之间的一个,country的格式为:xxx省(xxx市)(xxx县/区)
                    info.setCountry("中国");
                    if (index == length - 1) {
                        info.setProvince(country); // 设置省份,格式列入: 广东省
                    } else {
                        // 格式为:广东省广州市
                        info.setProvince(country.substring(0, index + 1)); // 设置省份
                        int index2 = country.indexOf('市', index); // 查看下一个出现市的位置
                        if (index2 > 0) {
                            country.substring(1,1);
                            info.setCity(country.substring(index + 1, Math.min(index2 + 1, length))); // 设置city
                        }
                    }
                } else {
                    // 其他的五个自治区 四个直辖市 2个特别行政区
                    String flag = country.substring(0, 2); // 拿字符串前两位
                    switch (flag) {
                    case "内蒙":
                        info.setCountry("中国");
                        info.setProvince("内蒙古自治区");
                        country = country.substring(3);
                        if (country != null && !country.isEmpty()) {
                            index = country.indexOf('市');
                            if (index > 0) {
                                info.setCity(country.substring(0, Math.min(index + 1, country.length()))); // 设置市
                            }
                        }
                        break;
                    case "广西":
                    case "西藏":
                    case "宁夏":
                    case "新疆":
                        info.setCountry("中国");
                        info.setProvince(flag);
                        country = country.substring(2);
                        if (country != null && !country.isEmpty()) {
                            index = country.indexOf('市');
                            if (index > 0) {
                                info.setCity(country.substring(0, Math.min(index + 1, country.length()))); // 设置市
                            }
                        }
                        break;
                    case "上海":
                    case "北京":
                    case "天津":
                    case "重庆":
                        info.setCountry("中国");
                        info.setProvince(flag + "市");
                        country = country.substring(3); // 去除这个省份/直辖市
                        if (country != null && !country.isEmpty()) {
                            index = country.indexOf('区');
                            if (index > 0) {
                                char ch = country.charAt(index - 1);
                                if (ch != '校' || ch != '小') {
                                    info.setCity(country.substring(0, Math.min(index + 1, country.length()))); // 设置区
                                }
                            }

                            if (RegionInfo.DEFAULT_VALUE.equals(info.getCity())) {
                                // city还是默认值
                                index = country.indexOf('县');
                                if (index > 0) {
                                    info.setCity(country.substring(0, Math.min(index + 1, country.length()))); // 设置区
                                }
                            }
                        }
                        break;
                    case "中国香港":
                    case "中国澳门":
                        info.setCountry("中国");
                        info.setProvince(flag + "特别行政区");
                        break;
                    default:
                        break;
                    }
                }
            }
        } catch (Exception e) {
            // 解析过程中出现异常
            e.printStackTrace();
        }
        return info;
    }

    /**
     * ip地域相关的一个model
     * 
     * @author gerry
     *
     */
    public static class RegionInfo {
        public static final String DEFAULT_VALUE = "unknown"; // 默认值
        private String country = DEFAULT_VALUE; // 国家
        private String province = DEFAULT_VALUE; // 省份
        private String city = DEFAULT_VALUE; // 城市

        public String getCountry() {
            return country;
        }

        public void setCountry(String country) {
            this.country = country;
        }

        public String getProvince() {
            return province;
        }

        public void setProvince(String province) {
            this.province = province;
        }

        public String getCity() {
            return city;
        }

        public void setCity(String city) {
            this.city = city;
        }

        @Override
        public String toString() {
            return "RegionInfo [country=" + country + ", province=" + province + ", city=" + city + "]";
        }
    }
}

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

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 2018-11-08 杀死If Else switch case(策略模式+工厂模式+map)套餐 Kill 项目中的switch case

    为了便于理解,举个没有业务逻辑的例子,基于这个例子上进行优化。 现在是12:47,举个饭后吃水果的例子哈哈哈(逃 假设我们可以选择的水果有香蕉、西瓜和苹果。吃香...

    Albert陈凯
  • Spark自定义累加器的实现

    Spark自定义累加器的实现 Java版本: package com.luoxuehuan.sparkproject.spark; import org.apa...

    Albert陈凯
  • 2019-04-30 自从用了Gitbook 简书用的都少了,来个水贴 MimeType OR MediaType

    Albert陈凯
  • 第50节:Java当中的泛型

    这就存在一个问题,如果集合存储元素时,而且存储对象有很多,而且对象类型不相同,就很容易导致隐患。

    达达前端
  • Kafka消费者分区分配策略及自定义分配策略

    我们知道kafka的主题中数据数据是按照分区的概念来的,一个主题可能分配了多个分区,每个分区配置了复制系数,为了可用性,在多个broker中进行复制,一个分区在...

    王知无
  • 几句简单代码解决springboot整合JdbcTemplate访问数据库进行操作

    这里只是一个简单的例子,但是基本的jdbc的操作是有了,具体的jdbcTemplate的使用可以查看

    好好学java
  • 打造Moya便捷解析库,提供RxSwift拓展

    LinXunFeng
  • 打造Moya便捷解析库,提供RxSwift拓展

    1、相信大家在使用Swift开发时,Moya是首选的网络工具,在模型解析这一块,Swift版模型解析的相关第三方库有很多,本人最习惯用的就是SwiftyJSON...

    LinXunFeng
  • Android入门之Http请求方式Get与Post

    前言:在Android开发的过程中,必须会接触到数据交互(访问数据,写入数据等你等),既然接触到数据的交互,那么自然而然就是使用通讯间的协议来进行请求,最常见的...

    AlicFeng
  • Java代码审计| Spring框架知识篇

    在上期的Java代码审计Spring框架思路篇中,斗哥为大家讲述了如何得到Spring审计的Demo,审计源码,根据Spring框架审计思路初步判定是否存在漏洞...

    漏斗社区

扫码关注云+社区

领取腾讯云代金券