您的位置:首页 > 运维架构

实战hadoop海量数据处理系列05 : 实现点击流日志的数据清洗模块

2017-06-10 21:02 399 查看

实战hadoop海量数据处理系列05 : 实现点击流日志的数据清洗模块

之前已经实现结构化数据的清洗,下一步我们将实现半结构化(非结构化)数据的清洗。

本文假设读者已搭建好了eclipse环境,并且已经导入ClickStreamETL文件夹下面的子工程。

如果遇到环境相关的问题,可以在专门的帖子下面留言。

在阅读本文前,强烈建议阅读原书“实现点击流日志的数据清洗模块”章节。

本文的代码同步于github,相关地址如下:

github地址

本系列博客专栏地址

overview

经典重现,引入原书流程图和map-reduce排序图说明原理。

为了还原案例,本项目增强了日志中的cookie,独有动态图演示执行过程,并在最后尾部给出开发中的若干思考。

全文脉络如下:

- 流程图

- 日志格式解析

- 排序流程图

- 重要代码解释

- 运行结果

- 工程实践思考

1 流程图

从原书摘取了经典的流程图,在于说明流程框架的重要性,因为流程贯穿全章节。



2日志格式解析

先列出一条日志,样式如下:

120.196.145.58 - - [11/Dec/2013:10:00:32 +0800] "GET /__utm.gif" 200 35 "http://easternmiles.ceair.com/flight/index.html" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/31.0.1650.63 Safari/537.36" "BIGipServermu_122.119.122.14=192575354.20480.0000;Webtrends=120.196.145.58.1386724976245806;uuid=sid1;userId=099;st=1"1482 352 - easternmiles.ceair.com 794


这里展示的日志,属于apache服务器日志,保留了原书字段,本项目在原有cookie
内部加入uuid,user id和sesstion time三个字段
,这些字段将在map阶段起重要作用。下面把里面cookie提取在此,其中uuid ,user id, session time分别是sid1,099,1;

"BIGipServermu_122.119.122.14=192575354.20480.0000;Webtrends=120.196.145.58.1386724976245806;uuid=sid1;userId=099;st=1"


完整的日志请见github中 log_3_2_1.txt文件,本文件将作为map reduce作业的输入。

2.1 日志中获取的相关字段

原书提供了非常详细字段列表信息,这里仅对
影响map reduce任务
的重要字段进行说明。

字段说明
ipaddressip地址, 可从点击流日志中获取
receive Time服务器接收时间, 可从点击流日志中获取
url由流日志中主机地址和请求的一行合成
unique id非重复的id,可从点击流的cookie中提取
session time会话发生的时间,可从点击流的cookie中提取
session id由unique id和session time合成

3 排序流程图

为了避免数据倾斜,作者对map的key进行了重构,利用unqiue id和session time合成了session id ,其中经典的流程图如下,一图胜过万千代码喔 ,已经熟悉本图的忠实读者可以略过。



4重要代码解释

代码主要分map/reduce/partion/二次排序等,

这里只贴出reducer,因为本项目对其进行了增强,其他代码请读者查看原有项目代码

4.1 Reducer部分

亮点在代码里,原先作者已经有很好的注释,改进的地方都用注释放在该行的尾部。

public class ClickStreamReducer extends Reducer<Text, Text, NullWritable, Text>{

///表示前一个sessionId
public static String preSessionId = "-";//66:改进版把他变为静态变量
static int csvp = 0;//66:改进版把csvp变为静态变量,并移到成员函数外面

protected void reduce(Text key, Iterable<Text> values, Context context) throws java.io.IOException ,InterruptedException {

String sessionId = key.toString().split("&")[0];

//如果是第一条数据
if(preSessionId.equals("-")){
csvp = 1;
preSessionId = sessionId;//66:改进版,新加入这句,不然逻辑不完整
}else{
//如果与前一个sessionId相同,说明是同一个session
if(preSessionId.equals(sessionId)){
//累加csvp
csvp++;
//如果不同,说明是新的session,重置preSessionId 和 csvp
}else{
preSessionId = sessionId;
csvp = 1;
}
}

//按照clickstream_log的格式再末尾加上csvp
String reduceOutValue = values.iterator().next().toString() + "\t" + csvp;
context.write(NullWritable.get(), new Text(reduceOutValue));
};
}


5 运行结果

5.1 map reduce结果

ip          uniqueId    sessionId   SessionTimesReceiveTime  UserId csvp          URL                               ReferUrl
120.196.145.58  sid1        sid1|1      1       1386727232000   099 1       easternmiles.ceair.com/__utm.gif   http://easternmiles.ceair.com/flight/index.html 120.196.145.58  sid1        sid1|1      1       1386727292000   099 2       easternmiles.ceair.com/__utm.gif   http://easternmiles.ceair.com/flight/index.html 120.196.145.58  sid1        sid1|1      1       1386727352000   099 3       easternmiles.ceair.com/__utm.gif   http://easternmiles.ceair.com/flight/index.html 120.196.145.58  sid2        sid2|10     10      1386727412000   199 1       easternmiles.ceair.com/__utm.gif   http://easternmiles.ceair.com/flight/index.html 120.196.145.58  sid2        sid2|10     10      1386727472000   199 2       easternmiles.ceair.com/__utm.gif   http://easternmiles.ceair.com/flight/index.html 120.196.145.58  sid3        sid3|100    100     1386727472000   299 1       easternmiles.ceair.com/__utm.gif   http://easternmiles.ceair.com/flight/index.html


从结果中可见,对于同一组seesion,按照接收时间(receive time)先后, 可见访问顺序(csvp)按照需要给予标记。

5.2 动态运行图

时长2分钟左右,请耐心观看。



6 小结

实践了作者的ClickStream作业,还原了过程例子演示的过程。

下一步计划,结合python调用jar文件完善开发过程。

7 其他 开发中的思考

7.1 处理正则表达式匹配过程的堆栈溢出

默认的堆栈大小在正则表达式匹配的时候不能满足内存需求,报以下错误

...
java.lang.Exception: java.lang.StackOverflowError at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:406)
Caused by: java.lang.StackOverflowError at java.util.regex.Pattern$CharProperty.match(Pattern.java:3692)
...


解决方法是在启动jvm时提大stack 的大小(这里本工程是使用-Xss40m)

7.2 处理DateFormat匹配的细节

在源代码(05-16)版本中,直接运行有以下问题:

17/06/10 09:36:18 INFO mapred.MapTask: record buffer = 262144/327680
java.text.ParseException: Unparseable date: "[11/Dec/2013:10:00:32 +0800]"
at java.text.DateFormat.parse(DateFormat.java:357)
at com.etl.mapreduce.ClickStreamMapper.map(ClickStreamMapper.java:84)


分析触发问题的代码,发现有两处,一是被处理的日期串还有额外的字符如[ ] ,第二是匹配串里面没有处理时区的问题,详细的解决方案见github相关代码ClickStreamMapper.java

7.3 调试工程经验

我的开发机是在windows, 部署机是在cent os ,因此存在远程调试的问题,该细节将会详细的总结为一章,请见后续更新
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息