图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Spark › 实操演练› Spark应用--日志处理实验之MapReduce方法
分享

Spark应用--日志处理实验之MapReduce方法

xioaxu790 2014-7-10 09:23:47 发表于 实操演练 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 1 14978
问题导读：
1、如何使用Spark处理日志数据？
2、日志数据都有什么不同之处？




附录：实验代码下载
1：创建日志格式处理类KPI
  1. package hadoop2.logs;

  2. 

  3. import java.text.ParseException;

  4. import java.text.SimpleDateFormat;

  5. import java.util.Date;

  6. import java.util.HashSet;

  7. import java.util.Locale;

  8. import java.util.Set;

  9. 

  10. /*

  11.  * KPI Object

  12.  */

  13. public class KPI {

  14.         private String remote_addr;// 记录客户端的ip地址

  15.         private String remote_user;// 记录客户端用户名称,忽略属性"-"

  16.         private String time_local;// 记录访问时间与时区

  17.         private String request;// 记录请求的url与http协议

  18.         private String status;// 记录请求状态；成功是200

  19.         private String body_bytes_sent;// 记录发送给客户端文件主体内容大小

  20.         private String http_referer;// 用来记录从那个页面链接访问过来的

  21.         private String http_user_agent;// 记录客户浏览器的相关信息

  22. 

  23.         private boolean valid = true;// 判断数据是否合法

  24. 

  25.         private static KPI parser(String line) {

  26.                 // System.out.println(line);

  27.                 KPI kpi = new KPI();

  28.                 String[] arr = line.split(" ");

  29.                 if (arr.length > 11) {

  30.                         kpi.setRemote_addr(arr[0]);

  31.                         kpi.setRemote_user(arr[1]);

  32.                         kpi.setTime_local(arr[3].substring(1));

  33.                         kpi.setRequest(arr[6]);

  34.                         kpi.setStatus(arr[8]);

  35.                         kpi.setBody_bytes_sent(arr[9]);

  36.                         kpi.setHttp_referer(arr[10]);

  37. 

  38.                         if (arr.length > 12) {

  39.                                 kpi.setHttp_user_agent(arr[11] + " " + arr[12]);

  40.                         } else {

  41.                                 kpi.setHttp_user_agent(arr[11]);

  42.                         }

  43. 

  44.                         if (Integer.parseInt(kpi.getStatus()) >= 400) {// 大于400，HTTP错误

  45.                                 kpi.setValid(false);

  46.                         }

  47.                 } else {

  48.                         kpi.setValid(false);

  49.                 }

  50.                 return kpi;

  51.         }

  52. 

  53.         /**

  54.          * 按page的pv分类

  55.          */

  56.         public static KPI filterPVs(String line) {

  57.                 return parser(line);

  58.                 // KPI kpi = parser(line);

  59.                 // Set<String> pages = new HashSet<String>();

  60.                 // pages.add("/about");

  61.                 // pages.add("/black-ip-list/");

  62.                 // pages.add("/cassandra-clustor/");

  63.                 // pages.add("/finance-rhive-repurchase/");

  64.                 // pages.add("/hadoop-family-roadmap/");

  65.                 // pages.add("/hadoop-hive-intro/");

  66.                 // pages.add("/hadoop-zookeeper-intro/");

  67.                 // pages.add("/hadoop-mahout-roadmap/");

  68.                 //

  69.                 // if (!pages.contains(kpi.getRequest())) {

  70.                 // kpi.setValid(false);

  71.                 // }

  72.                 // return kpi;

  73.         }

  74. 

  75.         /**

  76.          * 按page的独立ip分类

  77.          */

  78.         public static KPI filterIPs(String line) {

  79.                 return parser(line);

  80.                 // KPI kpi = parser(line);

  81.                 // Set<String> pages = new HashSet<String>();

  82.                 // pages.add("/about");

  83.                 // pages.add("/black-ip-list/");

  84.                 // pages.add("/cassandra-clustor/");

  85.                 // pages.add("/finance-rhive-repurchase/");

  86.                 // pages.add("/hadoop-family-roadmap/");

  87.                 // pages.add("/hadoop-hive-intro/");

  88.                 // pages.add("/hadoop-zookeeper-intro/");

  89.                 // pages.add("/hadoop-mahout-roadmap/");

  90.                 //

  91.                 // if (!pages.contains(kpi.getRequest())) {

  92.                 // kpi.setValid(false);

  93.                 // }

  94.                 //

  95.                 // return kpi;

  96.         }

  97. 

  98.         /**

  99.          * PV按浏览器分类

  100.          */

  101.         public static KPI filterBroswer(String line) {

  102.                 return parser(line);

  103.         }

  104. 

  105.         /**

  106.          * PV按小时分类

  107.          */

  108.         public static KPI filterTime(String line) {

  109.                 return parser(line);

  110.         }

  111. 

  112.         /**

  113.          * PV按访问域名分类

  114.          */

  115.         public static KPI filterDomain(String line) {

  116.                 return parser(line);

  117.         }

  118. 

  119.         @Override

  120.         public String toString() {

  121.                 StringBuilder sb = new StringBuilder();

  122.                 sb.append("valid:" + this.valid);

  123.                 sb.append("\nremote_addr:" + this.remote_addr);

  124.                 sb.append("\nremote_user:" + this.remote_user);

  125.                 sb.append("\ntime_local:" + this.time_local);

  126.                 sb.append("\nrequest:" + this.request);

  127.                 sb.append("\nstatus:" + this.status);

  128.                 sb.append("\nbody_bytes_sent:" + this.body_bytes_sent);

  129.                 sb.append("\nhttp_referer:" + this.http_referer);

  130.                 sb.append("\nhttp_user_agent:" + this.http_user_agent);

  131.                 return sb.toString();

  132.         }

  133. 

  134.         public String getRemote_addr() {

  135.                 return remote_addr;

  136.         }

  137. 

  138.         public void setRemote_addr(String remote_addr) {

  139.                 this.remote_addr = remote_addr;

  140.         }

  141. 

  142.         public String getRemote_user() {

  143.                 return remote_user;

  144.         }

  145. 

  146.         public void setRemote_user(String remote_user) {

  147.                 this.remote_user = remote_user;

  148.         }

  149. 

  150.         public String getTime_local() {

  151.                 return time_local;

  152.         }

  153. 

  154.         public Date getTime_local_Date() throws ParseException {

  155.                 SimpleDateFormat df = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss",

  156.                                 Locale.US);

  157.                 return df.parse(this.time_local);

  158.         }

  159. 

  160.         public String getTime_local_Date_hour() throws ParseException {

  161.                 SimpleDateFormat df = new SimpleDateFormat("yyyyMMddHH");

  162.                 return df.format(this.getTime_local_Date());

  163.         }

  164. 

  165.         public void setTime_local(String time_local) {

  166.                 this.time_local = time_local;

  167.         }

  168. 

  169.         public String getRequest() {

  170.                 return request;

  171.         }

  172. 

  173.         public String getRequest_domain() {

  174.                 // String str = this.request.replace(""", "").replace("http://", "")

  175.                 // .replace("https://", "");

  176.                 // return str.lastIndexOf("/") > 0 ? str.substring(0, str.lastIndexOf("/")) : "/";

  177.                 String rtnString="";                

  178.                 String[] request_domain = request.split("/");

  179. 

  180.                 if (request_domain.length > 3) {

  181.                         for (int i = 0; i < 3; i++) {

  182.                                 rtnString= rtnString + request_domain[i]+"/" ;

  183.                         }

  184.                 } else {

  185.                         for (int i = 0; i < request_domain.length; i++) {

  186.                                 rtnString= request.lastIndexOf("/") > 0 ? request.substring(0, request.lastIndexOf("/")) +"/" : "/";

  187.                         }

  188.                 }

  189.                 

  190.                 return rtnString;

  191.                 // string value = "192.168.128.33";

  192.                 // string[] names = value.split("\\.");

  193.                 // for (int i = 0; i < names.length; i++) {

  194.                 // system.out.println(names[i]);

  195. 

  196.         }

  197. 

  198.         public void setRequest(String request) {

  199.                 this.request = request;

  200.         }

  201. 

  202.         public String getStatus() {

  203.                 return status;

  204.         }

  205. 

  206.         public void setStatus(String status) {

  207.                 this.status = status;

  208.         }

  209. 

  210.         public String getBody_bytes_sent() {

  211.                 return body_bytes_sent;

  212.         }

  213. 

  214.         public void setBody_bytes_sent(String body_bytes_sent) {

  215.                 this.body_bytes_sent = body_bytes_sent;

  216.         }

  217. 

  218.         public String getHttp_referer() {

  219.                 return http_referer;

  220.         }

  221. 

  222.         public String getHttp_referer_domain() {

  223.                 if (http_referer.length() < 8) {

  224.                         return http_referer;

  225.                 }

  226. 

  227.                 String str = this.http_referer.replace(""", "").replace("http://", "")

  228.                                 .replace("https://", "");

  229.                 return str.indexOf("/") > 0 ? str.substring(0, str.indexOf("/")) : str;

  230.         }

  231. 

  232.         public void setHttp_referer(String http_referer) {

  233.                 this.http_referer = http_referer;

  234.         }

  235. 

  236.         public String getHttp_user_agent() {

  237.                 return http_user_agent;

  238.         }

  239. 

  240.         public void setHttp_user_agent(String http_user_agent) {

  241.                 this.http_user_agent = http_user_agent;

  242.         }

  243. 

  244.         public boolean isValid() {

  245.                 return valid;

  246.         }

  247. 

  248.         public void setValid(boolean valid) {

  249.                 this.valid = valid;

  250.         }

  251. 

  252.         public static void main(String args[]) {

  253.                 String line = "222.68.172.190 - - [18/Sep/2013:06:49:57 +0000] "GET /stru.zip HTTP/1.1" 200 19939 "http://www.angularjs.cn/A00n" "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/29.0.1547.66 Safari/537.36"";

  254.                 System.out.println(line);

  255.                 KPI kpi = new KPI();

  256.                 String[] arr = line.split(" ");

  257. 

  258.                 kpi.setRemote_addr(arr[0]);

  259.                 kpi.setRemote_user(arr[1]);

  260.                 kpi.setTime_local(arr[3].substring(1));

  261.                 kpi.setRequest(arr[6]);

  262.                 kpi.setStatus(arr[8]);

  263.                 kpi.setBody_bytes_sent(arr[9]);

  264.                 kpi.setHttp_referer(arr[10]);

  265.                 kpi.setHttp_user_agent(arr[11] + " " + arr[12]);

  266.                 // System.out.println(kpi);

  267.                  System.out.println(kpi.getRequest_domain());

  268. 

  269.                 try {

  270.                         SimpleDateFormat df = new SimpleDateFormat("yyyy.MM.dd:HH:mm:ss",

  271.                                         Locale.US);

  272.                         System.out.println(df.format(kpi.getTime_local_Date()));

  273.                         System.out.println(kpi.getTime_local_Date_hour());

  274.                         System.out.println(kpi.getHttp_referer_domain());

  275.                 } catch (ParseException e) {

  276.                         e.printStackTrace();

  277.                 }

  278.         }

  279. 

  280. }
复制代码


2：统计(所有日志)独立 ip 数目，即不同 ip 的总数
下面代码输出了独立IP的访问次数，同时通过一个reduce计数器ReportTest.TotalIP来得到不同IP的总数。由于例程中combine和reduce采用同一方法，所以在统计IP 的总数的时候，需要取消combine过程，不然计数器将是combine过程计数和reduce过程计数过程之和。当然如果combine和reduce用不同的方法，那就不必取消combine过程了。

  1. package hadoop2.logs;

  2. 

  3. import java.io.IOException;

  4. 

  5. import org.apache.hadoop.conf.Configuration;

  6. import org.apache.hadoop.fs.Path;

  7. import org.apache.hadoop.io.IntWritable;

  8. import org.apache.hadoop.io.Text;

  9. import org.apache.hadoop.mapreduce.Job;

  10. import org.apache.hadoop.mapreduce.Mapper;

  11. import org.apache.hadoop.mapreduce.Reducer;

  12. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

  13. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

  14. import org.apache.hadoop.util.GenericOptionsParser;

  15. 

  16. public class KPIIP {

  17.         

  18.         static enum ReportTest{

  19.                 TotalIP

  20.         }

  21.         public static class KPIIPMapper extends

  22.                         Mapper<Object, Text, Text, IntWritable> {

  23.                 private final static IntWritable one = new IntWritable(1);

  24.                 private Text word = new Text();

  25. 

  26.                 @Override

  27.                 public void map(Object key, Text value, Context context)

  28.                                 throws IOException, InterruptedException {

  29.                         KPI kpi = KPI.filterPVs(value.toString());

  30.                         word.set(kpi.getRemote_addr());

  31.                         context.write(word, one);

  32.                 }

  33.         }

  34. 

  35.         public static class KPIIPReducer extends

  36.                         Reducer<Text, IntWritable, Text, IntWritable> {

  37.                 private IntWritable result = new IntWritable();

  38. 

  39.                 @Override

  40.                 public void reduce(Text key, Iterable<IntWritable> values,

  41.                                 Context context) throws IOException, InterruptedException {

  42.                         int sum = 0;

  43.                         for (IntWritable val : values) {

  44.                                 sum += val.get();

  45.                         }

  46.                         result.set(sum);

  47.                         context.getCounter(ReportTest.TotalIP).increment(1);

  48.                         context.write(key, result);

  49.                 }

  50.         }

  51. 

  52.         public static void main(String[] args) throws Exception {

  53.                 Configuration conf = new Configuration();

  54.                 String[] otherArgs = new GenericOptionsParser(conf, args)

  55.                                 .getRemainingArgs();

  56.                 if (otherArgs.length != 2) {

  57.                         System.err.println("Usage: KPIIP <in> <out>");

  58.                         System.exit(2);

  59.                 }

  60. 

  61.                 Job job = new Job(conf, "KPIIP");

  62.                 job.setJarByClass(KPIIP.class);

  63.                 job.setMapperClass(KPIIPMapper.class);

  64.                 //job.setCombinerClass(KPIIPReducer.class);

  65.                 job.setReducerClass(KPIIPReducer.class);

  66.                 job.setOutputKeyClass(Text.class);

  67.                 job.setOutputValueClass(IntWritable.class);

  68.                 FileInputFormat.addInputPath(job, new Path(otherArgs[0]));

  69.                 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

  70.                 System.exit(job.waitForCompletion(true) ? 0 : 1);

  71.         }

  72. 

  73. }
复制代码




3：统计(所有日志)每个子目录访问次数
对于日志来说，客户端的访问还有很多种形式，有可能是点击的、有可能是链接等等，日志中记录下来的访问请求要根据需要进行处理来满足分析的需求，可以通过编写KPI.filterPVs方法来达到过滤的目的，用KPI.getRequest_domain来格式化要分析的用户请求数据，最终通过mapreduce来完成分析之目的。

  1. package hadoop2.logs;

  2. 

  3. import java.io.IOException;

  4. 

  5. import org.apache.hadoop.conf.Configuration;

  6. import org.apache.hadoop.fs.Path;

  7. import org.apache.hadoop.io.IntWritable;

  8. import org.apache.hadoop.io.Text;

  9. import org.apache.hadoop.mapreduce.Job;

  10. import org.apache.hadoop.mapreduce.Mapper;

  11. import org.apache.hadoop.mapreduce.Reducer;

  12. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

  13. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

  14. import org.apache.hadoop.util.GenericOptionsParser;

  15. 

  16. public class KPIPV {

  17. 

  18.         public static class KPIPVMapper extends

  19.                         Mapper<Object, Text, Text, IntWritable> {

  20.                 private final static IntWritable one = new IntWritable(1);

  21.                 private Text word = new Text();

  22. 

  23.                 @Override

  24.                 public void map(Object key, Text value, Context context)

  25.                                 throws IOException, InterruptedException {

  26.                         KPI kpi = KPI.filterPVs(value.toString());

  27.                         word.set(kpi.getRequest_domain());

  28.                         context.write(word, one);

  29.                 }

  30.         }

  31. 

  32.         public static class KPIPVReducer extends

  33.                         Reducer<Text, IntWritable, Text, IntWritable> {

  34.                 private IntWritable result = new IntWritable();

  35. 

  36.                 @Override

  37.                 public void reduce(Text key, Iterable<IntWritable> values,

  38.                                 Context context) throws IOException, InterruptedException {

  39.                         int sum = 0;

  40.                         for (IntWritable val : values) {

  41.                                 sum += val.get();

  42.                         }

  43.                         result.set(sum);

  44.                         context.write(key, result);

  45.                 }

  46.         }

  47. 

  48.         public static void main(String[] args) throws Exception {

  49.                 Configuration conf = new Configuration();

  50.                 String[] otherArgs = new GenericOptionsParser(conf, args)

  51.                                 .getRemainingArgs();

  52.                 if (otherArgs.length != 2) {

  53.                         System.err.println("Usage: KPIPV <in> <out>");

  54.                         System.exit(2);

  55.                 }

  56. 

  57.                 Job job = new Job(conf, "KPIPV");

  58.                 job.setJarByClass(KPIPV.class);

  59.                 job.setMapperClass(KPIPVMapper.class);

  60.                 job.setCombinerClass(KPIPVReducer.class);

  61.                 job.setReducerClass(KPIPVReducer.class);

  62.                 job.setOutputKeyClass(Text.class);

  63.                 job.setOutputValueClass(IntWritable.class);

  64.                 FileInputFormat.addInputPath(job, new Path(otherArgs[0]));

  65.                 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

  66.                 System.exit(job.waitForCompletion(true) ? 0 : 1);

  67.         }

  68. 

  69. }
复制代码


4：统计(所有日志)每个 ip，访问的子目录次数和上面的例子一样，只是map的key中多了一个IP信息。

  1. package hadoop2.logs;

  2. 

  3. import java.io.IOException;

  4. 

  5. import org.apache.hadoop.conf.Configuration;

  6. import org.apache.hadoop.fs.Path;

  7. import org.apache.hadoop.io.IntWritable;

  8. import org.apache.hadoop.io.Text;

  9. import org.apache.hadoop.mapreduce.Job;

  10. import org.apache.hadoop.mapreduce.Mapper;

  11. import org.apache.hadoop.mapreduce.Reducer;

  12. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

  13. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

  14. import org.apache.hadoop.util.GenericOptionsParser;

  15. 

  16. public class KPI3 {

  17.         public static class KPI3Mapper extends

  18.                         Mapper<Object, Text, Text, IntWritable> {

  19.                 private final static IntWritable one = new IntWritable(1);

  20.                 private Text word = new Text();

  21. 

  22.                 @Override

  23.                 public void map(Object key, Text value, Context context)

  24.                                 throws IOException, InterruptedException {

  25.                         KPI kpi = KPI.filterPVs(value.toString());

  26.                         word.set(kpi.getRemote_addr() + " " + kpi.getRequest_domain());

  27.                         context.write(word, one);

  28.                 }

  29.         }

  30. 

  31.         public static class KPI3Reducer extends

  32.                         Reducer<Text, IntWritable, Text, IntWritable> {

  33.                 private IntWritable result = new IntWritable();

  34. 

  35.                 @Override

  36.                 public void reduce(Text key, Iterable<IntWritable> values,

  37.                                 Context context) throws IOException, InterruptedException {

  38.                         int sum = 0;

  39.                         for (IntWritable val : values) {

  40.                                 sum += val.get();

  41.                         }

  42.                         result.set(sum);

  43.                         context.write(key, result);

  44.                 }

  45.         }

  46. 

  47.         public static void main(String[] args) throws Exception {

  48.                 Configuration conf = new Configuration();

  49.                 String[] otherArgs = new GenericOptionsParser(conf, args)

  50.                                 .getRemainingArgs();

  51.                 if (otherArgs.length != 2) {

  52.                         System.err.println("Usage: KPI3 <in> <out>");

  53.                         System.exit(2);

  54.                 }

  55. 

  56.                 Job job = new Job(conf, "KPI3");

  57.                 job.setJarByClass(KPI3.class);

  58.                 job.setMapperClass(KPI3Mapper.class);

  59.                 job.setCombinerClass(KPI3Reducer.class);

  60.                 job.setReducerClass(KPI3Reducer.class);

  61.                 job.setOutputKeyClass(Text.class);

  62.                 job.setOutputValueClass(IntWritable.class);

  63.                 FileInputFormat.addInputPath(job, new Path(otherArgs[0]));

  64.                 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

  65.                 System.exit(job.waitForCompletion(true) ? 0 : 1);

  66.         }

  67. 

  68. }

复制代码






关注公众号，获取大数据、人工智能20套、区块链资源5阶段等资源，随时更新，获取最新技术资源
回复

使用道具 举报

自定义广告语

已有(1)人评论

跳转到指定楼层
正序浏览
梦回三国 发表于 2014-11-6 10:11:20
这篇文章应该放到hadoop中吧，怎么在Spark中呢？？
回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
xioaxu790

金牌会员

关注

  • 851

    主题

  • 1000

    帖子

  • 108

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000