码迷,mamicode.com
首页 > 其他好文 > 详细

项目分析:日志分析

时间:2018-05-07 20:54:45      阅读:244      评论:0      收藏:0      [点我收藏+]

标签:arc   日志分析   保存   bcf   也会   logs   html   search   分隔符   

日志分析
目的:
生产中会生成大量的系统日志、应用程序日志、安全日志等日志,通过对日志的分析可以了解数据库的负载、健康状况,可以分析客户的分布情况、客户的行为,甚至基于这些分析可以做出预测。
 
一般流程:
日志产出-->采集(Logstash、Flume、Scribe)-->存储(落地,可不落地直接分析)-->分析-->存储(数据库、nosql)[持久化结果并方便查询]-->可视化(例如浏览器显示)
 
开源实时日志分析ELK平台
Logstash收集日志,并存放到ElasticSearch集群中,kibana则从ES集群中查询数据生成图标,返回浏览器端。
 
分析的前提:数据为半结构化数据。
日志是半结构化数据,是有组织的,有格式的数据,可以分割成行和列,就可以当做表理解和处理了,当然也可以分析里面的数据。
 
结构化数据:mysql内
半结构化:日志,文本文件。
非结构化:视频,音频。
 
文本分析:
日志是文本文件,需要依赖文件IO,字符串操作、正则表达式等技术。
通过这些技术把日志中需要的数据提取出来。
实例:分析以下日志,例行:
183.60.212.153 - - [19/Feb/2013:10:23:29 +0800] "GET /o2o/media.html?menu=3 HTTP/1.1" 200 16691 "-" "Mozilla/5.0 (compatible; EasouSpider; +http://www.easou.com/search/spider.html)"
一、数据提取
1)字符串切割:利用切片实现一次遍历使用多种分隔符分割:
def makekey(line:str):
    chars = set(‘ \t‘)
    start = 0
    skip = False
    for i,v in enumerate(line):
        if not skip and v in ‘"[‘:
            skip = True
            start = i +1
        elif skip and v in ‘"]‘:
            skip = False
            yield line[start:i]
            start = i+1
            continue
        elif skip:
            continue
        if not skip:
            if v in chars:
                if i == start:
                    start = i+1
                    continue
                yield line[start:i]
                start = i+1
    else:
        if start <len(line):
            yield line[start:]
 
print(list(makekey(line)))
2)正则表达式实现字符串切割
pattern =‘‘‘(?P<remote>[\d.]{7,}) - - \[(?P<datatime>[\w /+:]+)\] \
"(?P<method>\w+) (?P<url>\S+) (?P<protocol>[\w/\d.]+)" \
(?P<status>\d+)(?P<length>\d+) .+ "(?P<useragent>.+)"‘‘‘
 
二、处理数据-->转化为需要的类型:
①时间处理:
datetime
时间差类型:timedelta
转为秒数:timedelta.total_seconds()
1、字符串格式转时间格式:
def turn(timestr):
    return datetime.datetime.strptime(timestr,‘%d/%b/%Y:%H:%M:%S %z‘)
time = ‘19/Feb/2013:10:23:29 +0800‘
print(turn(time),type(turn(time)))
2、创建带时区的now:[带时区时间不能与不带时区时间运算]
tz = datetime.timezone(datetime.timedelta(hours=8))
current = datetime.datetime.now(tz)
3.datetime相减结果为timedelta对象,转为seconde需要total_seconds()方法
(current - start).total_seconds() >=interval
②map,zip实现三元组结合对应处理组成字典:
dict(map(lambda item:(item[0],item[2](item[1])),zip(name,makekey(line),ops)))
③使用命名分组提取需要的数据,并组成dict。调用使用匹配结果的groupdict:
pattern=‘abc’
regex = re.compile(pattern,str)
matcher = regex.match(line)
matcher.groupdict()
#####代码实现,切片makekey,构建数据字典####
name = (‘remote‘,‘‘,‘‘,‘datetime‘,‘request‘,‘status‘,‘lenth‘,‘‘,‘useragent‘)
ops = (None,None,None,lambda timestr:datetime.datetime.strptime(timestr,‘%d/%b/%Y:%H:%M:%S %z‘),\
       lambda request:dict(zip((‘method‘,‘url‘,‘proto‘),request.split())),int,int,None,None)
 
def extract(line:str):
    t=zip(name,makekey(line),ops)
    d=dict(map(lambda item:(item[0],item[2](item[1]) if item[2] is not None else item[1]),t))
    return d
print(extract(line))
############正则表达式命名元组实现########################
ops ={‘datatime‘:lambda timestr:datetime.datetime.strptime(timestr,‘%d/%b/%Y:%H:%M:%S %z‘),‘status‘:int,‘length‘:int}
 
regex = re.compile(pattern)
 
def extract(line:str) ->dict:
    matcher = regex.match(line)
    d={}
    for k,v in matcher.groupdict().items():
        d[k]=ops.get(k,lambda x:x)(v)  #在ops中找不到就返回一个什么都不做的函数。
    return d
 
print(extract(line))
三、数据载入
本项目是N行记录,载入数据就是文件IO。
def load(path):
    with open(path) as f:
        for line in f:
            ret = extract(line)
            if ret:
                yield ret
            else:
                continue  #解析失败就跳过
 
四、时间窗口分析【也叫滑动窗口】
很多数据,例如日志,都是和时间相关的,都是按照时间顺序产生的。
产生的数据分析的时候,要按照时间求值。
 
interval表示每一次求值的时间间隔。
width 时间窗口宽度,指的一次求值的时间窗口宽度。即对多长时间内的数据进行分析。
 
1、width > interval ,生产中最长用到的方式。
例如width=8,interval=5,操作为求平均数。即每5秒对近8秒的数据进行一次求平均数,数据求值时会有重叠使用。
技术分享图片
2.width = interval,例如:w=5,i=5. 每5秒对近5秒的数操作。数据求值时没有重叠使用。
技术分享图片
 
 
3.width<interval。一般不使用,会有采集到的数据不参与计算。
 
时序数据
运维环境中,日志、监控等产生的数据都是与时间相关的数据,按照时间先后产生并记录下来的数据,所以一般按照时间对数据进行分析。
 
数据分析基本程序结构
 
④map函数取出多字典相同key的value。
items=[{‘name‘:hh,value:10},{‘name‘:tt,value:20},{‘name‘:vv,value:30}]
list(map(lambda x:x[‘value‘],items)) 取出[10,20,30]  
 
无限生成随机数,产生时间相关的数据,返回时间和随机数的字典。
每次取3个数据,求平均值。
############模拟每生成3个数据后计算平均值#############################
def source():
    while True:
        yield{‘value‘:random.randint(1,10),‘datetime‘:datetime.datetime.now()}
 
s = source()
items = [next(s) for _ in range(3)]
 
def handle(iterable):
    mapper=map(lambda x:x[‘value‘],iterable)
    ret =sum(mapper)/len(iterable)
    return ret
 
print(‘{:.2f}‘.format(handle(items)))
##################################################
窗口函数的实现,每一段时间后计算平均值
################################################
def source(second=1):
    while True:
        yield {‘value‘:random.randint(1,10),‘datetime‘:datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8)))} #创建带时区的时间对象
        time.sleep(second)
 
def handle(iterable):
    return sum(map(lambda x : x[‘value‘],iterable))/len(iterable)
 
def window(iterator,handle,width:int,interval:int):
    """
    窗口函数
    :param iterator: 数据源,生成器,输送数据
    :param handle:    数据处理函数
    :param interval:  时间窗口宽度,秒
    :param width:   处理时间间隔,秒
    :return:    返回这段时间的平均值
    """
#不知道第一条的时间,设置一个比第一条时间早很多的时间,读入后替换掉。
    start = datetime.datetime.strptime(‘20170101 000000 +0800‘,‘%Y%m%d %H%M%S %z‘)
    current = datetime.datetime.strptime(‘20170101 010000 +0800‘,‘%Y%m%d %H%M%S %z‘)
    # start = datetime.datetime.strptime(‘20170101 000000‘,‘%Y%m%d %H%M%S‘)
    # current = datetime.datetime.strptime(‘20170101 010000‘,‘%Y%m%d %H%M%S‘)
    buffer = [] #窗口中待计算的数据
    delta = datetime.timedelta(seconds=width-interval)
    while True:
        #从数据源获取数据
        data = next(iterator)
        if data:
            buffer.append(data) #存入临时缓冲等待计算
            current = data[‘datetime‘]
 
        #每隔interval计算buffer中的数据一次
        if (current - start).total_seconds() >= interval:
            ret = handle(buffer)
            print(‘{:.2f}‘.format(ret))
            start = current
            #清除超出width的数据
            buffer = [x for x in buffer if x[‘datetime‘] > current -delta]
 
window(source(),handle,10,5)
#####################################################################
五、分发
生产者消费者模型
对于一个监控系统,需要处理很多数据,包括日志。对其中已有数据的采集、分析。
被监控对象就是数据的生产者producer,数据的处理程序就是数据的消费者consumer。
 
生产者消费者传统模型
技术分享图片
代码耦合太高,如果生成规模扩大,不易扩展,生产和消费的速度很难匹配等。
生产者和消费者的问题是什么?
举例:
卖包子的,如果包子卖不完,还要继续蒸包子,会怎么样?包子堆积。
如果先蒸一些,快卖完了再蒸一些。不会有等包子的队伍。
如果供不应求,还没和面,包子都被预定了,出现排队等包子情况。
综上、最核心的问题,就是生产者和消费者速度匹配的问题。
但是,往往速度是不能够很好的匹配。
解决的办法--队列queue
 
队列queue
作用---解耦、缓冲
技术分享图片
日志生产者往往会部署好几个程序,日志的产生也很多,而消费者也会有多个程序,去提取日志分
析处理。
 
数据的生产时不稳定的,会造成短时间数据的“潮涌”,需要缓冲。
消费者消费的能力不一样,有快有慢,消费者可以自己决定消费缓冲区中的数据。
 
单机可以使用queue内建的模块构建进程内的队列,满足多个线程间的生产消费需求。
大型系统可以使用第三方消息中间件:RabbitMQ、RocketMQ、kafka
 
queue模块--队列
queue模块提供了一个先进先出的队列Queue。队列是不可迭代的。
 
queue.Queue(maxsize=0)
创建FIFO队列,返回Queue对象。maxsize小于等于0,队列长度没有限制。
 
Queue.get(block=True,timeout=None)
从队列中溢出元素并返回这个元素
block为阻塞,timeout为超时。
1.如果block为True,是阻塞,timeout为None就是一直阻塞。
2.如果block为True但是timeout有值,就阻塞到一定秒数抛出Empty异常。
3.如果block为False,是非阻塞,timeout将被忽略,要么成功返回一个元素,要么抛出empty异常。
 
Queue.put(item,block=True,timeout=None)
把一个元素加入到队列中去。
block=True,timeout=None,一直阻塞到有空位放元素。
block=True,timeout=5,阻塞5秒就抛Full异常。
block=False,timeout失效,立刻返回,能塞进去就塞,不能则抛出Full异常。
 
Queue.put_nowait(item)
等价于put(item,False),也就是能塞进去就塞,不能则抛出Full异常。
 
⑤队列使用:
from queue import Queue
 
q = Queue()
q.put(10)
q.put(5)
 
print(q.get())  #10,先进先出
print(q.get())  #5
print(q.get(timeout=3))#阻塞,超时后抛queue.Empty异常。
####################################################################
分发器的实现:
生产者(数据源)生产数据,缓冲到消息队列中
 
数据处理流程:
数据加载-->提取-->分析(滑动窗口函数)
 
处理大量数据的时候,对于一个数据源来说,需要多个消费者处理。但是如何分配数据就是个问题了。
需要一个分发器(调度器),把数据分发给不同的消费者处理。
 
注册:消费者要想得到数据,必须在分发器注册,说明自己需要被分发数据。
分发:共用一个消息队列是可以的,但是需要解决争抢问题。此例中我们假设handle1,handle2为两
  个不同的分析函数,需要各自用用一个队列。
注册实现:在调度器内部记录那些消费者,每个消费者拥有自己的队列。
分发实现:线程,由于一条数据会被多个不同的注册过的handle处理,所以最好的方式多线程。
⑥线程使用举例
技术分享图片
分发器代码实现:
##################################################
def dispatcher(src):
    #分发器中记录handle,同时保存各自的队列
    handles = []    #记录消费者列表
    queues = []     #记录各自的队列
 
    def reg(handle,width:int,interval:int):
        """
        注册窗口处理函数
        每注册一个函数,需要为函数开辟一个独立队列,
        并开启一个独立线程运行它。
 
        """
        q = Queue()  #建立队列
        queues.append(q)
 
        h = threading.Thread(target=window,args = (q,handle,width,interval)) #线程
        handles.append(h)
 
    def run():
            for t in handles:
                t.start()
 
            for item in src:
                for q in queues:
                    q.put(item)
    return reg,run
 
reg,run = dispatcher(source())
 
reg(handle,10,5)
run()
##################################################################################

项目分析:日志分析

标签:arc   日志分析   保存   bcf   也会   logs   html   search   分隔符   

原文地址:https://www.cnblogs.com/flysck/p/9004631.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!