用Python构建基于Hadoop的MapReduce日志分析平台

jieforest

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

复制代码

jieforest

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

jieforest

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

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

复制代码

jieforest

详细的参数，对于咱们来说提供性能可以把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所占的域数目

jieforest

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

jieforest

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

可以写成shell文件的模式；

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

复制代码

jieforest

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

复制代码

jieforest

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

1. 13/12/14 13:26:52 INFO streaming.StreamJob: Tracking URL: http://101.rui.com:50030/jobdetails.jsp?jobid=job_201312131904_0030
   
2. 13/12/14 13:26:53 INFO streaming.StreamJob:  map 0%  reduce 0%
   
3. 13/12/14 13:27:16 INFO streaming.StreamJob:  map 100%  reduce 100%
   
4. 13/12/14 13:27:16 INFO streaming.StreamJob: To kill this job, run:
   
5. 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
   
6. 13/12/14 13:27:16 INFO streaming.StreamJob: Tracking URL: http://101.rui.com:50030/jobdetails.jsp?jobid=job_201312131904_0030
   
7. 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
   
8. 13/12/14 13:27:16 INFO streaming.StreamJob: killJob...
   
9. Streaming Command Failed!
   

复制代码

jieforest

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

jieforest

over.