项目需求:
需要统计一下线上日志中某些信息每天出现的频率,举个简单的例子,统计线上每天的请求总数和异常请求数。线上大概几十台
服务器,每台服务器大概每天产生4到5G左右的日志,假设有30台,每台5G的,一天产生的日志总量为150G。
处理方案:
方案1:传统的处理方式,写个JAVA日志分析代码,部署到每台服务器进行处理,这种方式部署起来耗时费力,又不好维护。
方案2:采用Hadoop分布式处理,日志分析是Hadoop集群系统的拿手好戏。150G每天的日志也算是比较大的数据量了,搭个简
单的Hadoop集群来处理这些日志是再好不过的了。
Hadoop集群的搭建:
参见这两篇文章:http://www.cnblogs.com/cstar/archive/2012/12/16/2820209.html
http://www.cnblogs.com/cstar/archive/2012/12/16/2820220.html
我们这里的集群就采用了两台机器,配置每台8核,32G内存,500G磁盘空间。
日志准备工作:
由于日志分散在各个服务器,所以我们先需要将所有的日志拷贝到我们的集群系统当中,这个可以通过linux服务器下rsync或者scp
服务来执行。这里我们通过scp服务来拷贝,由于都是内网的机器,所以拷贝几个G的日志可以很快就完成。下面是拷贝日志的脚本,脚本
还是有一些需要注意的地方,我们只需要拷贝前一天的数据,实际保存的数据可能是好几天的,所以我们只要把我们需要的这一天的数据
SCP过去就可以了。
#!/bin/sh workdir=/home/myproj/bin/log/ files=`ls $workdir` pre1date=`date + "%Y%m%d" -d "-1 days" `
pre1date1=`date + "%Y-%m-%d" -d "-1 days" `
curdate=`date + "%Y%m%d" `
hostname=`uname -n` echo $pre1date $curdate uploadpath= "/home/hadoop/hadoop/mytest/log/" $pre1date1 "/" $hostname
echo $uploadpath cd $workdir mintime= 240000
secondmintime= 0
for file in $files; do
filedate=`stat $file | grep Modify| awk '{print $2}' |sed -e 's/-//g' `
filetime=`stat $file | grep Modify| awk '{print $3}' |cut -d "." -f1 | sed -e 's/://g' | sed 's/^0\+//' `
if [ $filedate -eq $curdate ]; then
if [ $filetime -lt $mintime ]; then
secondmintime=$mintime
mintime=$filetime
fi
fi
done echo "mintime:" $mintime
step= 1000
mintime=`expr $mintime + $step` echo "mintime+1000:" $mintime
for file in $files; do
filedate=`stat $file | grep Modify| awk '{print $2}' |sed -e 's/-//g' `
filetime=`stat $file | grep Modify| awk '{print $3}' |cut -d "." -f1 | sed -e 's/://g' | sed 's/^0\+//' `
filename=`echo $file | cut -c 1 - 8 `
startchars= "info.log"
#echo $filename
if [ $filename == $startchars ]; then
if [ $filedate -eq $pre1date ]; then
scp -rp $file dir @antix2 :$uploadpath
#echo $file
elif [ $filedate -eq $curdate ]; then
if [ $filetime -lt $mintime ]; then
scp -rp $file dir @antix2 :$uploadpath
#echo $file
fi
fi
fi
#echo $filedate $filetime
done |
MapReduce代码
接下来就是编写MapReduce的代码了。使用Eclipse环境来编写,需要安装hadoop插件,我们hadoop机器采用的是1.1.1版本,所以插
件使用hadoop-eclipse-plugin-1.1.1.jar,将插件拷贝到eclipse的plugins目录下就可以了。然后新建一个MapReduce项目:
工程新建好了然后我们就可以编写我们的MapReduce代码了。
import java.io.IOException; import java.text.SimpleDateFormat; import java.util.Date; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class LogAnalysis { public static class LogMapper extends Mapper<LongWritable, Text, Text, IntWritable>{ private final static IntWritable one = new IntWritable(1); private Text word = new Text(); private Text hourWord = new Text(); public void map(LongWritable key, Text value, Context context ) throws IOException, InterruptedException { String line = value.toString(); SimpleDateFormat formatter2 = new SimpleDateFormat("yy-MM-dd"); java.util.Date d1 =new Date(); d1.setTime(System.currentTimeMillis()-1*24*3600*1000); String strDate =formatter2.format(d1); if(line.contains(strDate)){ String[] strArr = line.split(","); int len = strArr[0].length(); String time = strArr[0].substring(1,len-1); String[] timeArr = time.split(":"); String strHour = timeArr[0]; String hour = strHour.substring(strHour.length()-2,strHour.length()); String hourKey = ""; if(line.contains("StartASocket")){ word.set("SocketCount"); context.write(word, one); hourKey = "SocketCount:" + hour; hourWord.set(hourKey); context.write(hourWord, one); word.clear(); hourWord.clear(); } if(line.contains("SocketException")){ word.set("SocketExceptionCount"); context.write(word, one); hourKey = "SocketExceptionCount:" + hour; hourWord.set(hourKey); context.write(hourWord, one); word.clear(); hourWord.clear(); }
} } public static class LogReducer extends Reducer<Text,IntWritable,Text,IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } public static int run(String[] args) throws Exception{ Configuration conf = new Configuration(); String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length != 2) { System.err.println("Usage: loganalysis <in> <out>"); System.exit(2); } FileSystem fileSys = FileSystem.get(conf); String inputPath = "input/" + args[0]; fileSys.copyFromLocalFile(new Path(args[0]), new Path(inputPath));//将本地文件系统的文件拷贝到HDFS中 Job job = new Job(conf, "loganalysis"); job.setJarByClass(LogAnalysis.class); job.setMapperClass(LogMapper.class); job.setCombinerClass(LogReducer.class); job.setReducerClass(LogReducer.class); // 设置输出类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(inputPath)); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); Date startTime = new Date(); System.out.println("Job started: " + startTime); int ret = job.waitForCompletion(true)? 0 : 1; fileSys.copyToLocalFile(new Path(otherArgs[1]), new Path(otherArgs[1])); fileSys.delete(new Path(inputPath), true); fileSys.delete(new Path(otherArgs[1]), true); Date end_time = new Date(); System.out.println("Job ended: " + end_time); System.out.println("The job took " + (end_time.getTime() - startTime.getTime()) /1000 + " seconds."); return ret; } public static void main(String[] args) { try { int ret = run(args); System.exit(ret); } catch (Exception e) { e.printStackTrace(); System.out.println(e.getMessage()); } } }
部署到Hadoop集群:
代码完成后测试没有问题后,部署到集群当中去执行,我们有几十台服务器,所以每台的服务器的日志当成一个任务来执行。
workdir="/home/hadoop/hadoop/mytest" cd $workdir pre1date=`date +"%Y-%m-%d" -d "-1 days"` servers=(mach1 mach2 mach3 ) for i in ${servers[@]};do inputPath="log/"$pre1date"/"$i outputPath="output/log/"$pre1date"/"$i echo $inputPath $outputPath echo "start job "$i" date:"`date` hadoop jar LogAnalysis.jar loganalysis $inputPath $outputPath echo "end job "$i" date:"`date` done
相关推荐
基于Hadoop网站流量日志数据分析系统 1、典型的离线流数据分析系统 2、技术分析 - Hadoop - nginx - flume - hive - mysql - springboot + mybatisplus+vcharts nginx + lua 日志文件埋点的 基于Hadoop网站流量...
本次要实践的数据日志来源于国内某技术学习论坛,该论坛由某培训机构主办,汇聚了众多技术学习者,每天都有人发帖、回帖。...两个日志文件,一共有200MB,符合大数据量级,可以作为推荐系统数据集和hadoop测试集。
各个日志文件进行分析获取知识 由于日志具有数据量大 不易读懂的特点 如果仅凭借管理员查看日志记录的手段 其中所蕴含 的有用信息也难以发现 分布式计算技术正好可以用来解决这一难题 阐述了 syslog 日志收集流程 ...
至此,我们通过Python网络爬虫手段进行数据抓取,将我们网站数据(2013-05-30,2013-05-31)保存为两个日志文件,由于文件大小超出我们一般的分析工具处理的范围,故借助Hadoop来完成本次的实践。 2. 总体设计 2.1 ...
3.1.2 元数据更新及日志写入情景分析 3.1.3 Checkpoint过程情景分析 3.1.4 元数据可靠性机制 3.1.5 元数据一致性机制 3.2 使用说明 第4章 Hadoop的Backup Node方案 4.1 Backup Node概述 4.1.1 系统架构 4.1.2 使用...
利用HQL语言以及Hadoop分布式文件系统(HDFS)和MapReduce编程模式对海量搜索日志进行分析处理, 对用户搜索行为进行了分析研究。对用户搜索行为中的查询热点主题、用户点击数和URL排名、查询会话的分析结果对于搜索...
Twitter 则使用Hadoop 存储微博数据、日志文件和其他中间数据等。在国内,Hadoop 同样也得到了许多公司的青睐,如百度主要将Hadoop 应用于日志分析和网页数据库的数据 挖掘;阿里巴巴则将Hadoop 用于商业数据的排序...
hadoop 搜索引擎 用户行为分析 基于hadoop的搜索引擎用户行为分析,采用分布式文 件系统和并行计算模型支撑海量日志文件的处理
Hadoop分析气象数据完整版源代码(含Hadoop的MapReduce代码和SSM框架) 《分布式》布置了一道小作业,这是作业的所有代码,里面包含了Hadoop的MapReduce代码、和SSM框架显示数据的代码
- -------- 不懂运行,下载完可以私聊问,可远程教学 该资源内项目源码是个人的毕设,代码都测试ok,都是运行成功后才上传资源,...下载后请首先打开README.md文件(如有),仅供学习参考, 切勿用于商业用途。 --------
实验手册——搜狗搜索日志分析系统,压缩包里面是pdf格式和word格式的。 实验手册——搜狗搜索日志分析系统,可以练习使用hadoop离线分析。练手的吧。
055 安装Hadoop Eclipse插件并使用Eclipse查看文件系统 056 使用Eclipse运行MyWordCount程序并解决PriviledgedActionException问题 057 MyWordCount 处理过程详解 058 MapReduce提交作业源码跟踪讲解 059 MR作业...
基于hadoop 的购物网站日志分析系统+源代码+文档说明 -------- 不懂运行,下载完可以私聊问,可远程教学 该资源内项目源码是个人的毕设,代码都测试ok,都是运行成功后才上传资源,答辩评审平均分达到96分,放心下载...
基于Hadoop+Storm的网络日志实时分析系统+源代码+文档说明 -------- 不懂运行,下载完可以私聊问,可远程教学 该资源内项目源码是个人的毕设,代码都测试ok,都是运行成功后才上传资源,答辩评审平均分达到96分,...
实践的数据日志来源于...至此,我们通过Python网络手段进行数据抓取,将我们网站数据(2013-05-30,2013-05-31)保存为两个日志文件,由于文件大小超出我们一般的分析工具处理的范围,故借助Hadoop来完成本次的实践。
Hadoop分布式文件系统;Hadoop的I/O、MapReduce应用程序开发;MapReduce的工作机制;MapReduce的类型和格式;MapReduce的特性;如何构建Hadoop集群,如何管理Hadoop;Pig简介;Hbase简介;Hive简介;ZooKeeper简介;...
人工智能-hadoop
hadoop 搜索引擎 用户行为分析 基于hadoop的搜索引擎用户行为分析,采用分布式文件系统和并行计算模型支撑海量日志文件的处理.
第3章 Hadoop分布式文件系统 HDFS的设计 HDFS的概念 数据块 namenode和datanode 命令行接口 基本文件系统操作 Hadoop文件系统 接口 Java接口 从Hadoop URL中读取数据 通过FileSystem API读取数据 写入数据 目录 查询...
本文主要深入研究了Hadoop平台下的HDFS分布式文件系统,并行编程模型MapReduce,分布式列型存储数据库HBase。 二、对聚类分析进行研究。聚类分析是数据挖掘中应用极为广泛的一类,本文主要研究了聚类分析的起源、定义和...