日志分析
生产中会出现大量的系统日志、应用程序日志，安全日志等，通过贵日志的分析可以了解服务器的负载，健康状况，可以分析客户的分布情况、客户的行为，甚至基于这些分析可以做出预测。
一般采集流程：
日志产出->采集(logstash、Flumen、Scribe)->存储->分析->存储(数据库、NoSQL)->可视化
开源实时日志分析ELK平台
Logstash收集日志，并存放到ElasticSearch集群中，Kibana则从ES集群中查询数据生成图表，返回游览器端
数据提取
半结构化数据
日志是半结构化数据，是有组织的、有格式的数据。可以分割成行和列，就可以当作表理解和处理了，当然也可以分析里面的数据。
文本分析
日志是文本文件，需要依赖文件IO、字符串操作、正则表达式等技术
通过这些技术就可以将日志需要的数据提取出来了
使用正则表达式
import res=”’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 res=”’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 res=”’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 res=”’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:%vps云服务器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 Noneprint(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 fields else: continue
日志文件的加载
目前实现的代码中，只能接受一个路径，修改为接受一批路径。
可以约定一个路径下文件的存放形式：
如果送来的是一批路径，就迭代其中路径。如果路径是一个普通的文件，就直接加载这个文件。如果路径是一个目录，就便利路径下所有指定类型的文件，每一个我呢见按照行处理，可以提供参数处理是否递归子目录 from pathlib import Pathdef 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 Pathimport datetimeimport repattern=”'(?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 fields else: continuefrom pathlib import Pathdef 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) 27230270
