Python第七周 学习笔记（1）

标签：学习笔记

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

• 一般采集流程

  • 日志产出 -> 采集（Logstash、Flume、Scribe）-> 存储 -> 分析 -> 存储（数据库、NoSQL）-> 可视化

  • 开源实时日志分析ELK平台
  • Logstash收集日志，并存放到ElasticSearch集群中，Kibana则从ES集群中查询数据生成图表，返回浏览器端

分析的前提

半结构化数据

• 日志是半结构化数据，是有组织的，有格式的数据。可以分割成行和列，就可以当作表理解和处理，分析里面的数据

文本分析

• 日志是文本文件，需要依赖文件IO、字符串操作、正则表达式等技术

• 通过这些技术就能够把日志中需要的数据提取出来

• 目标数据形如：

  
  123.125.71.36 - - [06/Apr/2017:18:09:25 +0800] "GET / HTTP/1.1" 200 8642 "-" "Mozilla/5.0 (compatible; Baiduspider/2.0; +http://www.baidu.com/search/spider.html)"
  124.```

• nginx、tomcat等WEB Server都会产生这样的日志

提取数据

一、分割

import datetime

line = ‘‘‘
123.125.71.36 - - [06/Apr/2017:18:09:25 +0800] "GET / HTTP/1.1" 200 8642 "-" "Mozilla/5.0 (compatible; Baiduspider/2.0; +http://www.baidu.com/search/spider.html)"
‘‘‘

CHARS = set(" \t")

def makekey(line: str):
    start = 0
    skip = False
    for i, c in enumerate(line):
        if not skip and c in ‘"[‘:
            start = i + 1
            skip = True
        elif skip and c in ‘"]‘:
            skip = False
            yield line[start:i]
            start = i + 1
            continue

        if skip:
            continue

        if c in CHARS:
            if start == i:
                start = i + 1
                continue
            yield line[start:i]
            start = i + 1

    else:
        if start < len(line):
            yield line[start:]

names = (‘remote‘, ‘‘, ‘‘, ‘datetime‘, ‘request‘, ‘status‘, ‘length‘, ‘‘, ‘useragent‘)

ops = (None, None, None, lambda timestr: datetime.datetime.strptime(timestr, ‘%d/%b/%Y:%H:%M:%S %z‘),
       lambda request: dict(zip([‘method‘, ‘url‘, ‘protocol‘], request.split())),
       int, int, None, None)

def extract(line: str):
    return dict(map(lambda item: (item[0], item[2](item[1]) if item[2] is not None else item[1]),
                    zip(names, makekey(line), ops)))

print(extract(line))

二、正则表达式分割

PATTERN = r‘‘‘(?P<ip>[\d.]{7,})\s-\s-\s\[(?P<datetime>[^\[\]]+)\]\s"(?P<method>[^"\s]+)\s(?P<url>[^"\s]+)\s(?P<protocol>[^"\s]+)"\s(?P<status>\d{3})\s(?P<size>\d+)\s"(?:.+)"\s"(?P<useragent>[^"]+)"‘‘‘
pattern = re.compile(PATTERN)
ops = {‘datetime‘: (lambda x: datetime.datetime.strptime(x, ‘%d/%b/%Y:%H:%M:%S %z‘)), ‘status‘: int, ‘size‘: int}

def extract(text):

    mat = pattern.match(text)

    return {k: ops.get(k, lambda x: x)(v) for k, v in mat.groupdict().items()}

异常处理

• 日志中不免会出现一些不匹配的行，需要处理
• 这里使用re.match方法，有可能匹配不上。所以要增加一个判断
• 采用抛出异常的方式，让调用者获得异常并自行处理

PATTERN = r‘‘‘(?P<ip>[\d.]{7,})\s-\s-\s\[(?P<datetime>[^\[\]]+)\]\s"(?P<method>[^"\s]+)\s(?P<url>[^"\s]+)\s(?P<protocol>[^"\s]+)"\s(?P<status>\d{3})\s(?P<size>\d+)\s"(?:.+)"\s"(?P<useragent>[^"]+)"‘‘‘
pattern = re.compile(PATTERN)

ops = {‘datetime‘: (lambda x: datetime.datetime.strptime(x, ‘%d/%b/%Y:%H:%M:%S %z‘)), ‘status‘: int, ‘size‘: int}

def extract(text) -> dict:

    mat = pattern.match(text)

    if mat:
        return {k: ops.get(k, lambda x: x)(v) for k, v in mat.groupdict().items()}
    else:
        raise Exception(‘No match‘)

• 或者返回一个特殊值

def extract(text) -> dict:

    mat = pattern.match(text)

    if mat:
        return {k: ops.get(k, lambda x: x)(v) for k, v in mat.groupdict().items()}
    else:
        return None

滑动窗口

数据载入

def load(path):
    with open(path) as f:
        for line in f:
            fields = extract(line)
            if fields:
                yield fields
            else:
                continue

时间窗口分析

概念

• 很多数据，例如日志，都是和时间相关的，都是按照时间顺序产生的

• 产生的数据分析的时候，要按照时间求值

• interval表示每一次求值的时间间隔
• width时间窗接口宽度，指一次求值的时间窗口宽度

当width > interval

• 数据求值时会有重叠

当width = interval

• 数据求值没有重叠

时序数据

• 运维环境中，日志、监控等产生的数据都是与时间相关的数据，按照时间先后产生并记录下来的数据，所以一般按照时间对数据进行分析

数据分析基本程序结构

• 无限的生成随机数函数，产生时间相关的数据，返回时间和随机数字典

import random
import datetime
import time

def source():
    while True:
        yield {‘value‘: random.randint(1, 100), ‘datetime‘: datetime.datetime.now()}
        time.sleep(1)

s = source()
items = [next(s) for _ in range(3)]

def handler(iterable):
    return sum(map(lambda item: item[‘value‘], iterable)) / len(iterable)

print(items)
print("{:.2f}".format(handler(items)))

窗口函数实现

• 将上面的程序扩展为window函数

import random
import datetime
import time

def source(second=1):
    while True:
        yield {‘value‘: random.randint(1, 100),
               ‘datetime‘: datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8)))}
        time.sleep(second)

def window(iterator, handler, width: int, interval: int):
    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‘)

    buffer = []
    delta = datetime.timedelta(seconds=width - interval)

    while True:
        data = next(iterator)
        if data:
            buffer.append(data)
            current = data[‘datetime‘]

        if (current - start).total_seconds() >= interval:
            ret = handler(buffer)
            print(‘{:.2f}‘.format(ret))
            start = current
            buffer = [x for x in buffer if x[‘datetime‘] > current - delta]

def handler(iterable):
    return sum(map(lambda item: item[‘value‘], iterable)) / len(iterable)

window(source(), handler, 10, 5)

分发

生产者消费者模型

• 对于一个监控系统，需要处理很多数据，包括日志。对其中已有数据的采集、分析。
• 被监控对象就是数据的生产者producer，数据的处理程序就是数据的消费者consumer
• 生产者消费者传统模型
  
• 传统的生产者消费者模型，生产者生产，消费者消费。但这种模型有问题

• 开发的代码耦合高，如果生成规模扩大，不易扩展，生产和消费的速度很难匹配等。

• 解决办法：队列
• 作用：解耦、缓冲
  
• 日志生产者往往会部署好几个程序，日志产生的也很多，而消费者也会有很多个程序，去提取日志分析处理
• 数据生产是不稳定的。会造成段时间数据的潮涌，需要缓冲
• 消费者消费能力不一样，有快有慢，消费者可以自己决定消费缓冲区中的数据
• 单机可以使用queue内建的模块构建进程内的队列，满足多个线程间的生产消费需要
• 大型系统可以使用第三方消息中间件：RabbitMQ、RocketMQ、Kafka

queue模块——队列

• queue.Queue(maxsize=0)

  • 创建FIFO队列，返回Queue对象
  • maxsize小于等于0，队列长度没有限制

• Queue.get(block=True,timeout=None)

  • 从队列中移除元素并返回这个元素
  • block 阻塞，timeout 超时
  • 如果block为True，是阻塞，timeout为None就是一直阻塞
  • 如果block为True但是timeout有值，就阻塞到一定秒数抛出异常
  • block为False，是非阻塞，timeout将被忽略，要么成功返回一个元素，要么抛出empty异常

• Queue.get_nowait()

  • 等价于get(False)

• Queue.put(item,block=True,timeout=None)

  • 把一个元素加入到队列中去
  • block=True，timeout=None,一直阻塞直至有空位放元素
  • block=True，timeout=5,阻塞5秒就抛出Full异常
  • block=True，timeout失效，立刻返回，,一直阻塞直至有空位放元素
• Queue.put_nowait(item)
  • 等价于put(item,False)

分发器实现

• 生产者（数据源）生产数据，缓冲到消息队列中

• 数据处理流程：

  数据加载 -> 提取 -> 分析（滑动窗口函数）

• 处理大量数据的时候，可能需要多个消费者处理
• 需要一个分发器（调度器），把数据分发给不同的消费者处理
• 每一个消费者拿到数据后，有自己的处理函数。所以要有一种注册机制

数据加载 -> 提取 -> 分发 -> 分析函数1&分析函数2

• 分析1和分析2可以是不同的handler、窗口宽度、间隔时间

• 暂时采用轮询策略，一对多的副本发送，一个数据通过分发器、发送到多个消费者

• 消息队列

  • 在生产者和消费者之间使用消息队列，那么所有消费者可以共有一个消息队列，或各自拥有一个消息队列
  • 公用一个消息队列需要解决争抢问题。每个消费者拥有一个队列较易实现

• 注册

  • 在调度器内部记录消费者，每一个消费者拥有自己的队列
• 线程
  • 由于一条数据会被多个不同的注册过的handler处理，所以最好的方式就是多线程

分发器代码实现

def dispatcher(src):
    reg_handler = []
    queues = []

    def reg(handler, width, interval):
        q = Queue()
        queues.append(q)

        thrd = threading.Thread(target=window, args=(q, handler, width, interval))

        reg_handler.append(thrd)

    def run():

        for i in reg_handler:
            i.start()

        for item in src:
            for q in queues:
                q.put(item)

    return reg, run

reg, run = dispatcher(load(‘test.log‘))

reg(handler, 10, 5)
run()

整合代码

• load函数就是从日志中提取合格的数据生成函数
• 它可以作为dispatcher函数的数据源

import re
from pathlib import Path
import datetime
import time
import threading
from queue import Queue
from user_agents import parse

PATTERN = r‘‘‘(?P<ip>[\d.]{7,})\s-\s-\s\[(?P<datetime>[^\[\]]+)\]\s"(?P<method>[^"\s]+)\s(?P<url>[^"\s]+)\s(?P<protocol>[^"\s]+)"\s(?P<status>\d{3})\s(?P<size>\d+)\s"(?:.+)"\s"(?P<useragent>[^"]+)"‘‘‘
pattern = re.compile(PATTERN)

def extract(text):
    ops = {‘datetime‘: (lambda x: datetime.datetime.strptime(x, ‘%d/%b/%Y:%H:%M:%S %z‘)), ‘status‘: int, ‘size‘: int,
           ‘useragent‘: lambda x: parse(x)}

    mat = pattern.match(text)

    return {k: ops.get(k, lambda x: x)(v) for k, v in mat.groupdict().items()}

def openfile(filename):
    with open(filename) as f:
        for text in f:
            fields = extract(text)
            time.sleep(2)
            if fields:
                yield fields
            else:
                continue

# producer
def load(*pathnames):
    for path in pathnames:
        pathname = Path(path)
        if not pathname.exists():
            continue

        if pathname.is_file():
            yield from openfile(pathname)

        elif pathname.is_dir():
            for filename in pathname.iterdir():
                if filename.is_file():
                    yield from openfile(filename)

def sum_size_handler(iterable):
    return sum(map(lambda x: x[‘size‘], iterable))

def status_handler(iterable):
    status = {}
    for dic in iterable:
        key = dic[‘status‘]
        status[key] = status.get(key, 0) + 1

    return {k: v / len(iterable) for k, v in status.items()}

d = {}

def ua_handler(iterable):
    ua_family = {}
    for item in iterable:
        val = item[‘useragent‘]
        key = (val.browser.family, val.browser.version_string)
        ua_family[key] = ua_family.get(key, 0) + 1
        d[key] = d.get(key, 0) + 1
    return ua_family, d

# consumer
def window(q: Queue, handler, width, interval):
    st_time = datetime.datetime.strptime(‘19700101 000000 +0800‘, ‘%Y%m%d %H%M%S %z‘)
    cur_time = datetime.datetime.strptime(‘19700101 010000 +0800‘, ‘%Y%m%d %H%M%S %z‘)

    buffer = []

    while True:
        # src = next(iterable)
        src = q.get()
        print(src)
        buffer.append(src)

        cur_time = src[‘datetime‘]

        if (cur_time - st_time).total_seconds() > interval:
            val = handler(buffer)
            st_time = cur_time
            b, d = val
            d = sorted(d.items(), key=lambda x: x[1], reverse=True)
            print(val)
            print(d)
        buffer = [x for x in buffer if x[‘datetime‘] > (cur_time - datetime.timedelta(seconds=width - interval))]

def dispatcher(src):
    reg_handler = []
    queues = []

    def reg(handler, width, interval):
        q = Queue()
        queues.append(q)

        thrd = threading.Thread(target=window, args=(q, handler, width, interval))

        reg_handler.append(thrd)

    def run():

        for i in reg_handler:
            i.start()

        for item in src:
            for q in queues:
                q.put(item)

    return reg, run

if __name__ == ‘__main__‘:
    import sys

    # path=sys.argv[1]
    path = ‘test.log‘

reg, run = dispatcher(load(‘test.log‘))

# reg(sum_size_handler, 20, 5)
# reg(status_handler, 20, 5)
reg(ua_handler, 20, 5)
run()

完成分析功能

• 分析日志很重要，通过海量数据分析就能够知道是否遭受了攻击，是否被爬取及爬取高峰期，是否有盗链等

状态码分析

• 状态码中包含了很多信息。例如
  • 304，服务器收到客户端提交的请求参数，发现资源未变化，要求浏览器使用静态资源的缓存
  • 404，服务器找不到大请求的资源
  • 304占比大，说明静态缓存效果明显。404占比大，说明网站出现了错误链接，或者尝试嗅探网站资源
  • 如果400、500占比突然增大，网站一定出了问题。

def status_handler(iterable):
    status = {}
    for dic in iterable:
        key = dic[‘status‘]
        status[key] = status.get(key, 0) + 1

    return {k: v / len(iterable) for k, v in status.items()}

浏览器分析

useragent

• 这里指的是，软件按照一定的格式向远端的服务器提供一个表示自己的字符串
• 在HTTP协议中，使用useragent字段传送这个字符串

浏览器选项中可以修改此设置

信息提取

安装

pip install pyyaml ua-parser user-agents

数据分析

d = {}

def ua_handler(iterable):
    ua_family = {}
    for item in iterable:
        val = item[‘useragent‘]
        key = (val.browser.family, val.browser.version_string)
        ua_family[key] = ua_family.get(key, 0) + 1
        d[key] = d.get(key, 0) + 1
    return ua_family, d

Python第七周 学习笔记（1）

标签：学习笔记

原文地址：http://blog.51cto.com/11281400/2113294