离线数据分析平台实战——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'
操作步骤
- 修改pom文件，添加hadoop和hbase依赖
- 添加LoggerUtil类,中间设计到EventLogConstant常量类和TimeUtil工具类 LoggerUtil主要作用就是解析日志，返回一个map对象 EventLogConstants主要作用就是描述hbase的event_logs表的信息(表名，列簇名，列名)以及日志收集的日志中的数据参数name。其实列名和name是一样的。 TimeUtil主要作用解析服务器时间以及定义rowkey中的timestamp时间戳格式。
- 编写mapper类和runner类
- 添加环境变量文件，core-site.xml hbase-site.xml log4j.properties 根据不同的运行情况，修改源码将修改后的源码放到代码中。
- 添加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，集群运行测试：
代码语言：javascript复制
本地需要知道提交的job是需要提交到集群上的，所以需要指定两个参数mapreduce.framework.name和yarn.resourcemanager.address，value分别为yarn和hh:8032即可，但是可能会出现异常信息，此时需要将参数mapreduce.app-submission.cross-platform设置为true。
参数设置：
代码语言：javascript复制
mapreduce.framework.name=yarn
yarn.resourcemanager.address=hh:8032
mapreduce.app-submission.cross-platform=true
异常：
代码语言：javascript复制
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测试：
代码语言：javascript复制
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;
}
}
代码语言：javascript复制
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 "]";
}
}
代码语言：javascript复制
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 "]";
}
}
}
