python日志分析

article/2025/8/20 23:35:34

日志分析

生产中会出现大量的系统日志、应用程序日志，安全日志等，通过贵日志的分析可以了解服务器的负载，健康状况，可以分析客户的分布情况、客户的行为，甚至基于这些分析可以做出预测。
一般采集流程：

日志产出->采集(logstash、Flumen、Scribe)->存储->分析->存储(数据库、NoSQL)->可视化

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

数据提取

半结构化数据
日志是半结构化数据，是有组织的、有格式的数据。可以分割成行和列，就可以当作表理解和处理了，当然也可以分析里面的数据。
文本分析
日志是文本文件，需要依赖文件IO、字符串操作、正则表达式等技术
通过这些技术就可以将日志需要的数据提取出来了
使用正则表达式

import re
s='''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)\"'''
pattern='''([\d.]{3,}) - - \[(.*)\] "(\w+) (\S+) (.*)" (\d+ \d+) "-" "(.*)"'''
regex=re.compile(pattern)
def extract(log:str):m=regex.match(log)if m:print(m.groups())
extract(s)

在这里插入图片描述
使用命名分组

import re
s='''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)\"'''
pattern='''(?P<ip>[\d.]{3,}) - - \[(?P<time>.*)\] "(\w+) (\S+) (.*)" (\d+ \d+) "-" "(?P<useragent>.*)"'''
regex=re.compile(pattern)
def extract(log:str):m=regex.match(log)if m:print(m.groups())print('ip :{}'.format(m.group('ip')),'time :{}'.format(m.group('time')),'useragent :{}'.format(m.group('useragent')),sep='    ')
extract(s)

在这里插入图片描述
使用上面的分组就能提取想要的所有的组，也就是我们想要的数据，为了方便可以使用命名分组
映射
对每一个字段命名，然后与值和类型转换的方法对应
最简单的方式，就是使用正则表达式分组

import re
s='''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)\"'''
pattern='''(?P<ip>[\d.]{3,}) - - \[(?P<time>.*)\] "(\w+) (\S+) (.*)" (\d+ \d+) "-" "(?P<useragent>.*)"'''
regex=re.compile(pattern)
#使用映射将得到的结果转换成需要的类型
con={'datetime':lambda time:datetime.datetime.strptime(time,"%d/%b/%Y:%H:%M:%S %z"),
}#这里举例只转换了时间
def c(long:str):m=regex.match(long)if m:return {k:con.get(k,lambda x: x)(v) for k,v in m.groupdict().items()}
print(c(s))

在这里插入图片描述

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

import re
s='''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)\"'''
pattern='''(?P<ip>[\d.]{3,}) - - \[(?P<time>.*)\] "(\w+) (\S+) (.*)" (\d+ \d+) "-" "(?P<useragent>.*)"'''
regex=re.compile(pattern)
con={'time':lambda time:datetime.datetime.strptime(time,"%d/%b/%Y:%H:%M:%S %z"),
}
def c(long:str):m=regex.match(long)if m:return {k:con.get(k,lambda x: x)(v) for k,v in m.groupdict().items()}#考虑匹配不到的情况else:#采用报错#raise Exception('No match. {}'.format(s))#采用返回特殊值return None
print(c(s))
def load(path):with open(path) as f:for line in f:

数据载入
对于本项目来说，数据就是日志的一行行记录，载入数据就是文件IO的读取，将获取的数据的防范封装成函数

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

日志文件的加载
目前实现的代码中，只能接受一个路径，修改为接受一批路径。
可以约定一个路径下文件的存放形式：

如果送来的是一批路径，就迭代其中路径。
如果路径是一个普通的文件，就直接加载这个文件。
如果路径是一个目录，就便利路径下所有指定类型的文件，每一个我呢见按照行处理，可以提供参数处理是否递归子目录

from pathlib import Path
def load(*paths,encoding='utf-8',ext='*.log',recursive=False):for x in paths:p=Path(x)#目录处理if p.is_dir():if isinstance（ext,str）:ext=[ext]else:ext=list(ext)for e in ext:files=p.rglob(e)if recursiversive else p.glob(e)yield from loadfile(str(file.absolute()),encoding=encodingoding)elif p.is_file():yield from loadfile(str(file.absolute()),encoding=encodingoding)

完整代码

from pathlib import Path
import datetime
import re
pattern='''(?P<ip>[\d.]{3,}) - - \[(?P<time>.*)\] "(\w+) (\S+) (.*)" (\d+ \d+) "-" "(?P<useragent>.*)"'''
regex=re.compile(pattern)
conversion={"datetime":lambda timestr:datetime.datetime.strptime(timestr,'%d%b%Y:%H:%M:%S %z')
}
def extract(logline:str)->dict:"""返回字段的字典，如果返回None说明匹配失败"""m=regex.match(logline)if m:return {k:conversion.get(k,lambda x :x)(v)for k,v in m.groupdict().items()}else:return None #或输出日志记录
def loadfile(filename:str,encoding='utf-8'):"""装载日志文件"""with open(filename,encoding=encoding) as f :for line in f:fields=extract(lien)if fields:yield fieldselse:continue
from pathlib import Path
def load(*paths,encoding='utf-8',ext='*.log',recursive=False):"""装载日志文件"""for x in paths:p=Path(x)#目录处理if p.is_dir(): #处理目录if isinstance（ext,str）:ext=[ext]else:ext=list(ext)for e in ext:files=p.rglob(e)if recursiversive else p.glob(e)yield from loadfile(str(file.absolute()),encoding=encodingoding)elif p.is_file():yield from loadfile(str(file.absolute()),encoding=encodingoding)