使用python构建基于hadoop的mapreduce日志分析平台

简介:

出处:http://rfyiamcool.blog.51cto.com/1030776/1340057

          


224356461.jpg





流量比较大的日志要是直接写入Hadoop对Namenode负载过大,所以入库前合并,可以把各个节点的日志凑并成一个文件写入HDFS。 根据情况定期合成,写入到hdfs里面。

咱们看看日志的大小,200G的dns日志文件,我压缩到了18G,要是用awk perl当然也可以,但是处理速度肯定没有分布式那样的给力。

230102727.jpg


Hadoop Streaming原理

mapper和reducer会从标准输入中读取用户数据,一行一行处理后发送给标准输出。Streaming工具会创建MapReduce作业,发送给各个tasktracker,同时监控整个作业的执行过程。


任何语言,只要是方便接收标准输入输出就可以做mapreduce~

再搞之前我们先简单测试下shell模拟mapreduce的性能速度~

234955396.jpg


看下他的结果,350M的文件用时35秒左右。

235045406.jpg


这是2G的日志文件,居然用了3分钟。 当然和我写的脚本也有问题,我们是模拟mapreduce的方式,而不是调用shell下牛逼的awk,gawk处理。


001056805.jpg

awk的速度 !果然很霸道,处理日志的时候,我也很喜欢用awk,只是学习的难度有点大,不像别的shell组件那么灵活简单。

000946258.jpg


这是官方的提供的两个demo ~

map.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
#!/usr/bin/env python
"" "A more advanced Mapper, using Python iterators and generators." ""
import sys
def read_input(file):
for line  in file:
# split the line into words
yield line.split()
def main(separator= '\t' ):
# input comes from STDIN (standard input)
data = read_input(sys.stdin)
for words  in data:
# write the results to STDOUT (standard output);
# what we output here will be the input  for the
# Reduce step, i.e. the input  for reducer.py
#
# tab-delimited; the trivial word count  is 1
for word  in words:
print  '%s%s%d' % (word, separator,  1 )
if __name__ ==  "__main__" :
main()


reduce.py的修改方式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
#!/usr/bin/env python
"" "A more advanced Reducer, using Python iterators and generators." ""
from itertools  import groupby
from operator  import itemgetter
import sys
def read_mapper_output(file, separator= '\t' ):
for line  in file:
yield line.rstrip().split(separator,  1 )
def main(separator= '\t' ):
# input comes from STDIN (standard input)
data = read_mapper_output(sys.stdin, separator=separator)
# groupby groups multiple word-count pairs by word,
# and creates an iterator that returns consecutive keys and their group:
#   current_word - string containing a word (the key)
#   group - iterator yielding all [ "<current_word>" "<count>" ] items
for current_word, group  in groupby(data, itemgetter( 0 )):
try :
total_count = sum( int (count)  for current_word, count  in group)
print  "%s%s%d" % (current_word, separator, total_count)
except ValueError:
# count was not a number, so silently discard  this item
pass
if __name__ ==  "__main__" :
main()


咱们再简单点:

1
2
3
4
5
6
7
#!/usr/bin/env python
import sys
for line  in sys.stdin:
line = line.strip()
words = line.split()
for word  in words:
print  '%s\t%s' % (word,  1 )


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
#!/usr/bin/env python
from operator  import itemgetter
import sys
current_word = None
current_count =  0
word = None
for line  in sys.stdin:
line = line.strip()
word, count = line.split( '\t' 1 )
try :
count =  int (count)
except ValueError:
continue
if current_word == word:
current_count += count
else :
if current_word:
print  '%s\t%s' % (current_word, current_count)
current_count = count
current_word = word
if current_word == word:
print  '%s\t%s' % (current_word, current_count)


咱们就简单模拟下数据,跑个测试


084336884.jpg



剩下就没啥了,在hadoop集群环境下,运行hadoop的steaming.jar组件,加入mapreduce的脚本,指定输出就行了.  下面的例子我用的是shell的成分。


1
2
3
4
5
[root@ 101 cron]#$HADOOP_HOME/bin/hadoop  jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar \
-input myInputDirs \
-output myOutputDir \
-mapper cat \
-reducer wc


详细的参数,对于咱们来说提供性能可以把tasks的任务数增加下,根据情况自己测试下,也别太高了,增加负担。

(1)-input:输入文件路径

(2)-output:输出文件路径

(3)-mapper:用户自己写的mapper程序,可以是可执行文件或者脚本

(4)-reducer:用户自己写的reducer程序,可以是可执行文件或者脚本

(5)-file:打包文件到提交的作业中,可以是mapper或者reducer要用的输入文件,如配置文件,字典等。

(6)-partitioner:用户自定义的partitioner程序

(7)-combiner:用户自定义的combiner程序(必须用java实现)

(8)-D:作业的一些属性(以前用的是-jonconf),具体有:
           1)mapred.map.tasks:map task数目
           2)mapred.reduce.tasks:reduce task数目
           3)stream.map.input.field.separator/stream.map.output.field.separator: map task输入/输出数
据的分隔符,默认均为\t。
            4)stream.num.map.output.key.fields:指定map task输出记录中key所占的域数目
            5)stream.reduce.input.field.separator/stream.reduce.output.field.separator:reduce task输入/输出数据的分隔符,默认均为\t。
            6)stream.num.reduce.output.key.fields:指定reduce task输出记录中key所占的域数目


这里是统计dns的日志文件有多少行 ~

114512821.jpg

在mapreduce作为参数的时候,不能用太多太复杂的shell语言,他不懂的~

可以写成shell文件的模式;

1
2
3
4
5
6
7
8
#! /bin/bash
while read LINE;  do
#   for word  in $LINE
#   do
#    echo  "$word 1"
awk  '{print $5}'
done
done


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#! /bin/bash
count= 0
started= 0
word= ""
while read LINE; do
goodk=`echo $LINE | cut -d  ' ' -f  1 `
if "x" == x "$goodk" ];then
continue
fi
if "$word" !=  "$goodk" ];then
[ $started -ne  0 ] && echo -e  "$word\t$count"
word=$goodk                                                                                                                  
count= 1
started= 1
else
count=$(( $count +  1 ))
fi
done


有时候会出现这样的问题,好好看看自己写的mapreduce程序 ~

13/12/14 13:26:52 INFO streaming.StreamJob: Tracking URL: http://101.rui.com:50030/jobdetails.jsp?jobid=job_201312131904_0030

13/12/14 13:26:53 INFO streaming.StreamJob:  map 0%  reduce 0%

13/12/14 13:27:16 INFO streaming.StreamJob:  map 100%  reduce 100%

13/12/14 13:27:16 INFO streaming.StreamJob: To kill this job, run:

13/12/14 13:27:16 INFO streaming.StreamJob: /usr/local/hadoop/libexec/../bin/hadoop job  -Dmapred.job.tracker=localhost:9001 -kill job_201312131904_0030

13/12/14 13:27:16 INFO streaming.StreamJob: Tracking URL: http://101.rui.com:50030/jobdetails.jsp?jobid=job_201312131904_0030

13/12/14 13:27:16 ERROR streaming.StreamJob: Job not successful. Error: # of failed Map Tasks exceeded allowed limit. FailedCount: 1. LastFailedTask: task_201312131904_0030_m_000000

13/12/14 13:27:16 INFO streaming.StreamJob: killJob...

Streaming Command Failed!


python做为mapreduce执行成功后,结果和日志一般是放在你指定的目录下的,结果是在part-00000文件里面~


172550144.jpg

本文转自大角牛博客51CTO博客,原文链接http://blog.51cto.com/jingshengsun888/1341123如需转载请自行联系原作者


运维的戏子

相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
相关文章
|
6天前
|
存储 监控 安全
《SelectDB 新一代日志存储分析平台解决方案》白皮书重磅发布|立即下载
作为基于 Apache Doris 打造的现代化数据仓库,SelectDB 不拘泥于传统数仓的限制,针对日志数据的特点引入了多项创新性技术,使用户可基于 SelectDB 构建开放、高性能、低成本、统一的日志存储分析平台, 截至目前已在近百家行业内知名企业中落地。
《SelectDB 新一代日志存储分析平台解决方案》白皮书重磅发布|立即下载
|
11天前
|
分布式计算 Hadoop Java
Hadoop MapReduce编程
该教程指导编写Hadoop MapReduce程序处理天气数据。任务包括计算每个城市ID的最高、最低气温、气温出现次数和平均气温。在读取数据时需忽略表头,且数据应为整数。教程中提供了环境变量设置、Java编译、jar包创建及MapReduce执行的步骤说明,但假设读者已具备基础操作技能。此外,还提到一个扩展练习,通过分区功能将具有相同尾数的数字分组到不同文件。
16 1
|
13天前
|
机器学习/深度学习 人工智能 DataWorks
人工智能平台PAI产品使用合集之在使用行调用时遇到一直卡在ps job的问题,并且无法在DataWorks上查看到相关日志,是什么导致的
阿里云人工智能平台PAI是一个功能强大、易于使用的AI开发平台,旨在降低AI开发门槛,加速创新,助力企业和开发者高效构建、部署和管理人工智能应用。其中包含了一系列相互协同的产品与服务,共同构成一个完整的人工智能开发与应用生态系统。以下是对PAI产品使用合集的概述,涵盖数据处理、模型开发、训练加速、模型部署及管理等多个环节。
|
18天前
|
数据采集 SQL 分布式计算
|
17天前
|
消息中间件 数据采集 关系型数据库
离线数仓(三)【业务日志采集平台搭建】(2)
离线数仓(三)【业务日志采集平台搭建】
|
17天前
|
消息中间件 前端开发 Kafka
离线数仓(二)【用户行为日志采集平台搭建】(1)
离线数仓(二)【用户行为日志采集平台搭建】
|
28天前
|
分布式计算 C语言 Python
基于Python实现MapReduce
一、什么是MapReduce 首先,将这个单词分解为Map、Reduce。 • Map阶段:在这个阶段,输入数据集被分割成小块,并由多个Map任务处理。每个Map任务将输入数据映射为一系列(key, value)对,并生成中间结果。 • Reduce阶段:在这个阶段,中间结果被重新分组和排序,以便相同key的中间结果被传递到同一个Reduce任务。每个Reduce任务将具有相同key的中间结果合并、计算,并生成最终的输出。
|
1月前
|
监控 应用服务中间件 nginx
使用 Docker Compose V2 快速搭建日志分析平台 ELK (Elasticsearch、Logstash 和 Kibana)
ELK的架构有多种,本篇分享使用的架构如图所示: Beats(Filebeat) -> -> Elasticsearch -> Kibana,目前生产环境一天几千万的日志,内存占用大概 10G
64 4
|
29天前
|
分布式计算 Hadoop Java
Hadoop MapReduce 调优参数
对于 Hadoop v3.1.3,针对三台4核4G服务器的MapReduce调优参数包括:`mapreduce.reduce.shuffle.parallelcopies`设为10以加速Shuffle,`mapreduce.reduce.shuffle.input.buffer.percent`和`mapreduce.reduce.shuffle.merge.percent`分别设为0.8以减少磁盘IO。
|
29天前
|
存储 监控 Kubernetes
构建高效稳定的云原生日志监控系统
【5月更文挑战第26天】 随着微服务架构和容器化技术的普及,传统的日志监控方法面临重大挑战。本文将探讨如何构建一个既高效又稳定的云原生日志监控系统,该系统旨在提供实时的日志分析能力,同时保证系统的高可用性和可扩展性。我们将讨论利用现代技术栈如Fluentd、Elasticsearch和Kibana(EFK栈)来搭建日志收集、存储和可视化的解决方案,并深入探讨如何通过容器编排工具如Kubernetes来实现日志服务的自动伸缩和故障恢复。此外,我们还将介绍一些最佳实践,帮助运维团队在保持系统性能的同时,降低资源消耗和运营成本。