文档章节

实时日志分析平台搭建笔记(二)

季牧云
 季牧云
发布于 2017/07/04 10:18
字数 704
阅读 59
收藏 1

    接上一篇。ES采用两台服务器的集群,创建一个indices,名称为access-log,jstorm和spark处理后的日志写入其中。

注意:创建完access-log后,在写入数据前,需要运行以下命令,"time"为时间字段,"vulType"为_type

curl -XPUT '192.168.32.32:9200/access-log?pretty' -H 'Content-Type: application/json' -d'
{
  "mappings": {
    "vulType": {
      "properties": {
        "time": {
          "type":   "date",
          "format": "date_time_no_millis"
        }
      }
    }
  }
}
'

伪造一些日志,可以在kibana中看到jstorm执行后的结果

以上是jstorm实时检测的流程。还有一路是spark通过MR分析多条日志,进行统计规则的检测。

spark首先去hbase中读取数据,随后进行Map/Reduce,统计出触发风险规则的日志。例如测试统计三分钟内访问次数超过10次的IP。

获取hbase中的accesslog

public List<String> getTableByDate(String startRow,String endRow){
		Scan s = new Scan();
		List<String> logList = new ArrayList<String>();
		s.setStartRow(Bytes.toBytes(startRow));
		s.setStopRow(Bytes.toBytes(endRow));
		try {
			ResultScanner resultScanner = table.getScanner(s);
			for(Result rs:resultScanner){
				NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> navigableMap = rs.getMap();
				for(Map.Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> entry:navigableMap.entrySet()){
	                NavigableMap<byte[], NavigableMap<Long, byte[]>> map =entry.getValue();
	                for(Map.Entry<byte[], NavigableMap<Long, byte[]>> en:map.entrySet()){
	                    NavigableMap<Long, byte[]> ma = en.getValue();
	                    for(Map.Entry<Long, byte[]>e: ma.entrySet()){
	                    	logList.add(Bytes.toString(e.getValue()));
	                    }
	                }
	            }
			}
		} catch (IOException e) {
			e.printStackTrace();
			return null;
		}
		return logList;

随后将accesslog映射到类AccessLog中

public void analysesLog(String startKey,String endKey){
		logList = hbaseOperator.getTableByDate(startKey, endKey);
		listLength = logList.size();
		accesslogList = new ArrayList<AccessLog>(listLength);
		String patternstr = "((\\d+\\.){3}\\d+)\\s(\\S+)\\s\\[(.+)\\]\\s\"((\\S+)\\s(.*)\\s(\\S+))\"\\s\"(.*)\"\\s(\\d+)\\s(\\d+)\\s(\\S+)\\s\"(.*)\"\\s\"(.*)\"";
		pattern = Pattern.compile(patternstr);
		for(int i=0;i<listLength;i++){
			m = pattern.matcher(logList.get(i));
			if(m.find()){
				AccessLog accessLog = new AccessLog();
				accessLog.setLogType("access-log");
				accessLog.setClientIp(m.group(1));
				accessLog.setClientUser(m.group(3));
				SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZZ");
				SimpleDateFormat sdf = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss ZZ",Locale.ENGLISH);
				Date time = null;
				try {
					time = sdf.parse(m.group(4));
					accessLog.setTime(sdf2.format(time));	
				} catch (ParseException e) {
					accessLog.setTime(sdf2.format(new Date()));
					//logger.error("[LogSplit.execute]:" + e.getMessage());
				}
				accessLog.setMethod(m.group(6));
				accessLog.setUrl(m.group(7));
				accessLog.setVersion(m.group(8));
				accessLog.setRequestBody(m.group(9));
				accessLog.setStatus(m.group(10));
				accessLog.setHttpBytes(m.group(11));
				accessLog.setRequestTime(m.group(12));
				accessLog.setReferer(m.group(13));
				accessLog.setUserAgent(m.group(14));
				accesslogList.add(accessLog);
			}
		}
		analysesIp();
		
		
	}

映射完成后调用analysesIp()进行Map/Reduce操作,并将命中的IP写入到ES中

public void analysesIp(){
		if(!accesslogList.isEmpty()){
			List<String> ipList = new ArrayList<String>(listLength);
			Iterator<AccessLog> iterator = accesslogList.iterator();
			while (iterator.hasNext()) {
				ipList.add(iterator.next().getClientIp());
			}
			JavaRDD<String> ipRdd = sparkContext.parallelize(ipList);
			JavaPairRDD<String, Integer> clientIpRdd = ipRdd.mapToPair(initCount);
			JavaPairRDD<String, Integer> sumRdd = clientIpRdd.reduceByKey(sum);
			Map<String, Integer> ipMap = sumRdd.collectAsMap();
			SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZZ");
			AccessLog accessLog = new AccessLog();
			accessLog.setLogType("accesslog");
			accessLog.setVulType("IP访问异常");
			accessLog.setVulTypeId(Integer.toString(RuleLength.START_RULE_LENGTH));
			for(Entry<String, Integer> entry:ipMap.entrySet()){
				if(entry.getValue() > 30){
					accessLog.setTime(sdf2.format(new Date()));
					accessLog.setClientIp(entry.getKey());
					accessLog.setMsg("源IP在3分钟内共访问了" + entry.getValue() + "次");
					elasticSearch.inputData(accessLog);
				}
			}
			
		}
	}

用到两个简单的map/reduce函数,一个进行初始化,将单个IP初始化一个元组(IP,1)

第二个将相同的IP进行累加,并记录出现的次数,累加后的效果为(IP1,10) (IP2,3)

private static PairFunction<String, String, Integer> initCount = new PairFunction<String, String, Integer>() {

		/**
		 * 
		 */
		private static final long serialVersionUID = -6290488020645730311L;

		public Tuple2<String, Integer> call(String x){
			return new Tuple2<String, Integer>(x, 1);
		}
	};
	
	private static Function2<Integer, Integer, Integer> sum = new Function2<Integer, Integer, Integer>() {
		
		/**
		 * 
		 */
		private static final long serialVersionUID = 391813718009018019L;

		@Override
		public Integer call(Integer x, Integer y) throws Exception {
			
			return x + y;
		}
	};

写入ES后,可看到的效果如下

© 著作权归作者所有

共有 人打赏支持
季牧云
粉丝 20
博文 26
码字总数 20788
作品 0
浦东
其他
私信 提问
基于Flume+Log4j+Kafka的日志采集架构方案

http://www.linuxidc.com/Linux/2016-05/131402.htm (一)kafka-jstorm集群实时日志分析 之 ---------kafka实时日志处理 ELK 实现 Java 分布式系统日志分析架构 ELK(ElasticSearch, Logstas...

cccyb
2016/12/12
42
0
Logstash+Redis+Elasticsearch+Kibana+Nginx搭建日志分析系统

前言: 随着实时分析技术的发展及成本的降低,用户已经不仅仅满足于离线分析。目前我们服务的用户包括微博、微盘、云存储、弹性计算平台等十多个部门的多个产品的日志搜索分析业务,每天处理...

小柒2012
2016/03/02
1K
0
2017年12月19-20日 阿里云 飞天/智能 云栖大会-北京峰会 会议笔记

2017年12月19-20日,在北京国家会议中心举办了云栖大会北京峰会,19日为Tech Insight,20日为主论坛和其他分论坛。场次很多,内容很丰富,自己的一些参会笔记整理如下: 20171219 上午 阿里云...

海洋的云
2017/12/21
0
0
Kafka是个奇葩!——Linkin论文学习笔记

是个消息中间件吗?那和市面上其他一堆堆的中间件例如ActiveMQ, RabbitMQ有什么区别? 答案只有一个: Kafka是个集群的消息中间件+存储,一个节点可以存储几T的数据! 为啥一个中间件需要存储...

难易
2014/08/26
0
9
袋鼠云日志助力云南某金融机构日志平台建设,实现核心业务系统运维智能化

一、“黑匣子”:我很小,但我很重要 今天聊日志之前,我们先来聊另外一个话题:“黑匣子” 。 没错,就是飞机的“黑匣子”。黑匣子是记载飞行数据和驾驶舱通话的设备,它可以记载飞机停止作...

袋鼠云
11/09
0
0

没有更多内容

加载失败,请刷新页面

加载更多

撬动世界的支点——《引爆点》读书笔记2900字优秀范文

撬动世界的支点——《引爆点》读书笔记2900字优秀范文: 作者:挽弓如月。因为加入火种协会的读书活动,最近我连续阅读了两本论述流行的大作,格拉德威尔的《引爆点》和乔纳伯杰的《疯传》。...

原创小博客
16分钟前
0
0
《配电网自动化技术》第一章

写了配电网的组成、历程、难点、存在问题、解决方案,还是蛮好的。尤其是各地建设的系统后续又无法实用化,以及多种终端反而增加了运维工作量等,都是目前切实存在的让大家不停吐槽的内容。

max佩恩
21分钟前
0
0

中国龙-扬科
39分钟前
2
0
使用vuex的state状态对象的5种方式

vuex是一个专门为vue.js设计的状态管理模式,并且也可以使用devtools进行调试。 下面给大家来贴一下我的vuex的结构 下面是store文件夹下的state.js和index.js内容 //state.jsconst state =...

peakedness丶
43分钟前
2
0
NetCore MVC Demo

地址:http://114.116.9.72:5411

whltian
50分钟前
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部