百木园大数据量文本文件高效解析方案代码实现
测试环境
Python 3.6.2
Win 10 内存 8G，CPU I5 1.6 GHz
背景描述
这个作品来源于一个日志解析工具的开发，这个开发过程中遇到的一个痛点，就是日志文件多，日志数据量大，解析耗时长。在这种情况下，寻思一种高效解析数据解析方案。
解决方案描述
1、采用多线程读取文件
2、采用按块读取文件替代按行读取文件
由于日志文件都是文本文件，需要读取其中每一行进行解析，所以一开始会很自然想到采用按行读取，后面发现合理配置下，按块读取，会比按行读取更高效。
按块读取来的问题就是，可能导致完整的数据行分散在不同数据块中，那怎么解决这个问题呢？解答如下：
将数据块按换行符\\n切分得到日志行列表，列表第一个元素可能是一个完整的日志行，也可能是上一个数据块末尾日志行的组成部分，列表最后一个元素可能是不完整的日志行（即下一个数据块开头日志行的组成部分），也可能是空字符串(日志块中的日志行数据全部是完整的)，根据这个规律，得出以下公式，通过该公式，可以得到一个新的数据块，对该数据块二次切分，可以得到数据完整的日志行
上一个日志块首部日志行 +\\n + 尾部日志行 + 下一个数据块首部日志行 + \\n + 尾部日志行 + ...
3、将数据解析操作拆分为可并行解析部分和不可并行解析部分
数据解析往往涉及一些不可并行的操作，比如数据求和，最值统计等，如果不进行拆分，并行解析时势必需要添加互斥锁，避免数据覆盖，这样就会大大降低执行的效率，特别是不可并行操作占比较大的情况下。
对数据解析操作进行拆分后，可并行解析操作部分不用加锁。考虑到Python GIL的问题，不可并行解析部分替换为单进程解析。
4、采用多进程解析替代多线程解析
采用多进程解析替代多线程解析，可以避开Python GIL全局解释锁带来的执行效率问题，从而提高解析效率。
5、采用队列实现“协同”效果
引入队列机制，实现一边读取日志，一边进行数据解析：
- 日志读取线程将日志块存储到队列，解析进程从队列获取已读取日志块，执行可并行解析操作
- 并行解析操作进程将解析后的结果存储到另一个队列，另一个解析进程从队列获取数据，执行不可并行解析操作。
代码实现
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import re
import time
from datetime import datetime
from joblib import Parallel, delayed, parallel_backend
from collections import deque
from multiprocessing import cpu_count
import threading
class LogParser(object):
def __init__(self, chunk_size=1024*1024*10, process_num_for_log_parsing=cpu_count()):
self.log_unparsed_queue = deque() # 用于存储未解析日志
self.log_line_parsed_queue = deque() # 用于存储已解析日志行
self.is_all_files_read = False # 标识是否已读取所有日志文件
self.process_num_for_log_parsing = process_num_for_log_parsing # 并发解析日志文件进程数
self.chunk_size = chunk_size # 每次读取日志的日志块大小
self.files_read_list = [] # 存放已读取日志文件
self.log_parsing_finished = False # 标识是否完成日志解析
def read_in_chunks(self, filePath, chunk_size=1024*1024):
\"\"\"
惰性函数（生成器），用于逐块读取文件。
默认区块大小：1M
\"\"\"
with open(filePath, \'r\', encoding=\'utf-8\') as f:
while True:
chunk_data = f.read(chunk_size)
if not chunk_data:
break
yield chunk_data
def read_log_file(self, logfile_path):
\'\'\'
读取日志文件
这里假设日志文件都是文本文件，按块读取后，可按换行符进行二次切分，以便获取行日志
\'\'\'
temp_list = [] # 二次切分后，头，尾行日志可能是不完整的，所以需要将日志块头尾行日志相连接，进行拼接
for chunk in self.read_in_chunks(logfile_path, self.chunk_size):
log_chunk = chunk.split(\'\\n\')
temp_list.extend([log_chunk[0], \'\\n\'])
temp_list.append(log_chunk[-1])
self.log_unparsed_queue.append(log_chunk[1:-1])
self.log_unparsed_queue.append(\'\'.join(temp_list).split(\'\\n\'))
self.files_read_list.remove(logfile_path)
def start_processes_for_log_parsing(self):
\'\'\'启动日志解析进程\'\'\'
with parallel_backend(\"multiprocessing\", n_jobs=self.process_num_for_log_parsing):
Parallel(require=\'sharedmem\')(delayed(self.parse_logs)() for i in range(self.process_num_for_log_parsing))
self.log_parsing_finished = True
def parse_logs(self):
\'\'\'解析日志\'\'\'
method_url_re_pattern = re.compile(\'(HEAD|POST|GET)\\s+([^\\s]+?)\\s+\',re.DOTALL)
url_time_taken_extractor = re.compile(\'HTTP/1\\.1.+\\|(.+)\\|\\d+\\|\', re.DOTALL)
while self.log_unparsed_queue or self.files_read_list:
if not self.log_unparsed_queue:
continue
log_line_list = self.log_unparsed_queue.popleft()
for log_line in log_line_list:
#### do something with log_line
if not log_line.strip():
continue
res = method_url_re_pattern.findall(log_line)
if not res:
print(\'日志未匹配到请求URL，已忽略：\\n%s\' % log_line)
continue
method = res[0][0]
url = res[0][1].split(\'?\')[0] # 去掉了 ?及后面的url参数
# 提取耗时
res = url_time_taken_extractor.findall(log_line)
if res:
time_taken = float(res[0])
else:
print(\'未从日志提取到请求耗时，已忽略日志：\\n%s\' % log_line)
continue
# 存储解析后的日志信息
self.log_line_parsed_queue.append({\'method\': method,
\'url\': url,
\'time_taken\': time_taken,
})
def collect_statistics(self):
\'\'\'收集统计数据\'\'\'
def _collect_statistics():
while self.log_line_parsed_queue or not self.log_parsing_finished:
if not self.log_line_parsed_queue:
continue
log_info = self.log_line_parsed_queue.popleft()
# do something with log_info
with parallel_backend(\"multiprocessing\", n_jobs=1):
Parallel()(delayed(_collect_statistics)() for i in range(1))
def run(self, file_path_list):
# 多线程读取日志文件
for file_path in file_path_list:
thread = threading.Thread(target=self.read_log_file,
name=\"read_log_file\",
args=(file_path,))
thread.start()
self.files_read_list.append(file_path)
# 启动日志解析进程
thread = threading.Thread(target=self.start_processes_for_log_parsing, name=\"start_processes_for_log_parsing\")
thread.start()
# 启动日志统计数据收集进程
thread = threading.Thread(target=self.collect_statistics, name=\"collect_statistics\")
thread.start()
start = datetime.now()
while threading.active_count() > 1:
print(\'程序正在努力解析日志...\')
time.sleep(0.5)
end = datetime.now()
print(\'解析完成\', \'start\', start, \'end\', end, \'耗时\', end - start)
if __name__ == \"__main__\":
log_parser = LogParser()
log_parser.run([\'access.log\', \'access2.log\'])
注意：
需要合理的配置单次读取文件数据块的大小，不能过大，或者过小，否则都可能会导致数据读取速度变慢。笔者实践环境下，发现10M~15M每次是一个比较高效的配置。
作者：授客
微信/QQ：1033553122
全国软件测试QQ交流群：7156436
Git地址：https://gitee.com/ishouke
友情提示：限于时间仓促，文中可能存在错误，欢迎指正、评论！
作者五行缺钱，如果觉得文章对您有帮助，请扫描下边的二维码打赏作者，金额随意，您的支持将是我继续创作的源动力，打赏后如有任何疑问，请联系我!!!
微信打赏
支付宝打赏 全国软件测试交流QQ群
来源：https://www.cnblogs.com/shouke/p/16975025.html
本站部分图文来源于网络，如有侵权请联系删除。