spark分析日志
2016-11-07 本文已影响251人
IamOnelong
import java.text.SimpleDateFormat;
import java.util.*;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import scala.Tuple2;
public class SparkDemo {
static final String USER = "huanglong";
public static void main(String[] args) throws Exception {
SparkConf conf =new SparkConf().setAppName("Spark Log").setMaster("local[4]");
JavaSparkContext sc = new JavaSparkContext(conf); //其底层就是scala的sparkcontext
//JavaSparkContext sc = new JavaSparkContext("local[4]", "Spark Log", "/user/huanglong/spark", new String[0], envs);
String file="file:///Users/huanglong/access_05_30.log";
JavaRDD<String> data = sc.textFile(file, 4).cache();
//日志格式
//27.19.74.143 - - [30/May/2013:17:38:20 +0800] "GET /static/image/common/faq.gif HTTP/1.1" 200 1127;
JavaRDD<String> filter = data.filter(new org.apache.spark.api.java.function.Function<String, Boolean>() {
@Override
public Boolean call(String s) throws Exception {
//过滤合法和非静态的资源的日志
return (s.contains("POST")||s.contains("GET"))&& !s.contains("/static/");
}
});
final SimpleDateFormat tf = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss Z",
Locale.ENGLISH);
//只有一天的记录，不需要年月日了,时间精度到分钟
final SimpleDateFormat tf1 = new SimpleDateFormat("HH:mm");
JavaRDD<Map<String,String>> lines = filter.flatMap(new FlatMapFunction<String,Map<String,String>>(){
@Override
public Iterator<Map<String,String>> call(String s) throws Exception {
List<Map<String,String>> list = new ArrayList<Map<String,String>>(1);
Map<String,String> map = new HashMap<String, String>();
list.add(map);
String ip = s.substring(0,s.indexOf(" - - "));
map.put("ip",ip);
String t = s.substring(s.indexOf("[")+1,s.indexOf("]"));
Date date = tf.parse(t);
map.put("time",tf1.format(date));
String http = s.substring(s.indexOf("\"")+1,s.indexOf("\" "));
String[] h= http.split(" ");
if(h.length>1){
if(h[1]!=null&&h[1].indexOf("?")!=-1)
{
map.put("url",h[1].substring(0,h[1].indexOf("?")));
}else {
map.put("url",h[1]);
}
}
if(h.length>2)
{
map.put("version",h[2]);
}
if(map.get("version")!=null&&map.get("version").contains("1.1"))
{
String state = s.substring(s.indexOf("1.1\" ")+5,s.length());
String[] strings = state.split(" ");
if(strings.length>0) {
map.put("state", strings[0]);
}
}else if(map.get("version")!=null&&map.get("version").contains("1.0"))
{
String state = s.substring(s.indexOf("1.0\" ")+5,s.length());
String[] strings = state.split(" ");
if(strings.length>0) {
map.put("state", strings[0]);
}
}else {
//System.out.println(s);
}
return list.iterator();
}
});
System.out.println("总记录数:"+lines.count());
// 统计ip
ipsHandle(lines);
System.out.println("--------------------------------------------------");
// 统计分时页面访问量
timeHandle(lines);
System.out.println("--------------------------------------------------");
// 统计分时ip访问量
timeHandle2(lines);
System.out.println("--------------------------------------------------");
// 统计状态 400，300 200 各有多少,计算失败率
stateHandle(lines);
System.out.println("--------------------------------------------------");
// 统计http版本 1.1、1.0
versionHandle(lines);
// 页面热度
System.out.println("--------------------------------------------------");
pageHandle(lines);
sc.close();
}
private static void pageHandle(JavaRDD<Map<String, String>> lines) {
JavaRDD<Map<String, String>> filterData = lines.filter(new Function<Map<String, String>, Boolean>() {
@Override
public Boolean call(Map<String, String> map) throws Exception {
String url = map.get("url");
return "200".equals(map.get("state"))&&url.endsWith(".php");
}
});
JavaPairRDD<String,Integer> versionPairs = filterData.mapToPair(new PairFunction<Map<String,String>, String,Integer>() {
@Override
public Tuple2<String,Integer> call(Map<String, String> stringStringMap) throws Exception {
return new Tuple2<String, Integer>(stringStringMap.get("url"),1);
}
});
JavaPairRDD<String,Integer> versionCount = versionPairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1+v2;
}
});
versionCount.foreach(new VoidFunction<Tuple2<String,Integer>>() {
@Override
public void call(Tuple2<String, Integer> pairs) throws Exception {
System.out.println("URL->PV："+pairs._1+" : " +pairs._2);
}
});
}
private static void versionHandle(JavaRDD<Map<String, String>> lines) {
JavaPairRDD<String,Integer> versionPairs = lines.mapToPair(new PairFunction<Map<String,String>, String,Integer>() {
@Override
public Tuple2<String,Integer> call(Map<String, String> stringStringMap) throws Exception {
return new Tuple2<String, Integer>(stringStringMap.get("version"),1);
}
});
JavaPairRDD<String,Integer> versionCount = versionPairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1+v2;
}
});
versionCount.foreach(new VoidFunction<Tuple2<String,Integer>>() {
@Override
public void call(Tuple2<String, Integer> pairs) throws Exception {
System.out.println("VERSION->PV："+pairs._1+" : " +pairs._2);
}
});
}
private static void stateHandle(JavaRDD<Map<String, String>> lines) {
JavaPairRDD<String,Integer> statePairs = lines.mapToPair(new PairFunction<Map<String,String>, String,Integer>() {
@Override
public Tuple2<String,Integer> call(Map<String, String> stringStringMap) throws Exception {
return new Tuple2<String, Integer>(stringStringMap.get("state"),1);
}
});
JavaPairRDD<String,Integer> stateCount = statePairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1+v2;
}
});
stateCount.foreach(new VoidFunction<Tuple2<String,Integer>>() {
@Override
public void call(Tuple2<String, Integer> pairs) throws Exception {
System.out.println("STATE->PV："+pairs._1+" : " +pairs._2);
}
});
}
private static void ipsHandle(JavaRDD<Map<String, String>> lines) {
JavaPairRDD<String,Integer> ipPairs = lines.mapToPair(new PairFunction<Map<String,String>, String,Integer>() {
@Override
public Tuple2<String,Integer> call(Map<String, String> stringStringMap) throws Exception {
return new Tuple2<String, Integer>(stringStringMap.get("ip"),1);
}
});
JavaPairRDD<String,Integer> ipCount = ipPairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1+v2;
}
});
System.out.println("总ip数:"+ipCount.count());
ipCount.foreach(new VoidFunction<Tuple2<String,Integer>>() {
@Override
public void call(Tuple2<String, Integer> pairs) throws Exception {
System.out.println("IP->PV："+pairs._1+" : " +pairs._2);
}
});
}
private static void timeHandle2(JavaRDD<Map<String, String>> lines) {
JavaPairRDD<String,Map<String,Boolean>> timePairs = lines.mapToPair(new PairFunction<Map<String,String>, String,Map<String,Boolean>>() {
@Override
public Tuple2<String,Map<String,Boolean>> call(Map<String, String> stringStringMap) throws Exception {
Map<String,Boolean> m = new HashMap<String, Boolean>();
m.put(stringStringMap.get("ip"),Boolean.TRUE);
return new Tuple2<String, Map<String,Boolean>>(stringStringMap.get("time"),m);
}
});
JavaPairRDD<String,Map<String,Boolean>> timeCount = timePairs.reduceByKey(new Function2<Map<String,Boolean>, Map<String,Boolean>, Map<String,Boolean>>() {
@Override
public Map<String,Boolean> call(Map<String,Boolean> v1, Map<String,Boolean> v2) throws Exception {
HashMap<String, Boolean> m = new HashMap<String, Boolean>();
m.putAll(v1);
m.putAll(v2);
return m;
}
});
timeCount.foreach(new VoidFunction<Tuple2<String,Map<String,Boolean>>>() {
@Override
public void call(Tuple2<String, Map<String,Boolean>> pairs) throws Exception {
System.out.println("TIME->IP："+pairs._1+" : " +pairs._2.size());
}
});
}
private static void timeHandle(JavaRDD<Map<String, String>> lines) {
JavaPairRDD<String,Integer> timePairs = lines.mapToPair(new PairFunction<Map<String,String>, String,Integer>() {
@Override
public Tuple2<String,Integer> call(Map<String, String> stringStringMap) throws Exception {
return new Tuple2<String, Integer>(stringStringMap.get("time"),1);
}
});
JavaPairRDD<String,Integer> timeCount = timePairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1+v2;
}
});
timeCount.foreach(new VoidFunction<Tuple2<String,Integer>>() {
@Override
public void call(Tuple2<String, Integer> pairs) throws Exception {
System.out.println("TIME->PV："+pairs._1+" : " +pairs._2);
}
});
}
}
