文档章节

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

季牧云
 季牧云
发布于 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后,可看到的效果如下

© 著作权归作者所有

共有 人打赏支持
季牧云
粉丝 21
博文 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
从京东618数据井喷看大数据平台峰值处理制胜关键

作者介绍 京东集团618作战指挥中心,成员来自于京东各个技术体系,包括核心系统架构师、一线运维专家、科研学者等。近200位成员在618时共同努力,确保流量洪峰来临时系统安全、稳定、可靠,致...

博文视点
09/14
0
0

没有更多内容

加载失败,请刷新页面

加载更多

计算卷积神经网络浮点数运算量

前言 本文主要是介绍了,给定一个卷积神经网络的配置之后,如何大概估算它的浮点数运算量。 相关代码:CalFlops,基于MXNet框架的 Scala 接口实现的一个计算MXNet网络模型运算量的demo。 正文...

Ldpe2G
今天
1
0
Sql语言与MySql数据库

1. 数据库简介 1. 数据库,就是存储数据的仓库,只能通过sql语言来访问,数据库也是一个文件系统。通常,MySQL、Oracle等数据库,也被称为关系型数据库,其保存的不仅仅只是数据,还包括数据...

江左煤郎
今天
1
0
IDEA 取消自动import .*

打开设置 > Editor > Code Style > Java > Scheme Default > Imports ① 将 Class count to use import with "*" 改为 99 (导入同一个包的类超过这个数值自动变为 * ) ② 将 Names count ......

乔老哥
今天
3
0
PostGIS学习笔记(开篇)

PostGIS事实上算是笔者开始写博客的第一篇内容。而事实上那篇博文的内容并不丰富,笔者对PostGIS的了解仍然不多,然而17年在OSGeo课程学习时对PostGIS又有了进一步了解,并逐步发现它的强大。...

胖胖雕
今天
3
0
【Centos】在nginx服务器中配置php和mysql

接上一章《【Centos】利用Vultr服务器和namesilo布网》(https://my.oschina.net/u/3776619/blog/2051986),在Centos中配置好nginx,并在iptables中开启了80端口,和为了远程mysql操作方便开...

yongh701
今天
7
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部