项目分析：日志分析

Posted 2020-11-03 flysck

日志分析

目的：

生产中会生成大量的系统日志、应用程序日志、安全日志等日志，通过对日志的分析可以了解数据库的负载、健康状况，可以分析客户的分布情况、客户的行为，甚至基于这些分析可以做出预测。

 

一般流程：

日志产出-->采集（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()

##################################################################################