当前位置:首页 > 技术 > 正文内容

Python 大数据量文本文件高效解析方案代码实现

Lotus2022-12-19 10:19技术

大数据量文本文件高效解析方案代码实现

测试环境

Python 3.6.2

Win 10 内存 8G,CPU I5 1.6 GHz

背景描述

这个作品来源于一个日志解析工具的开发,这个开发过程中遇到的一个痛点,就是日志文件多,日志数据量大,解析耗时长。在这种情况下,寻思一种高效解析数据解析方案。

解决方案描述

1、采用多线程读取文件

2、采用按块读取文件替代按行读取文件

由于日志文件都是文本文件,需要读取其中每一行进行解析,所以一开始会很自然想到采用按行读取,后面发现合理配置下,按块读取,会比按行读取更高效。

按块读取来的问题就是,可能导致完整的数据行分散在不同数据块中,那怎么解决这个问题呢?解答如下:

将数据块按换行符\n切分得到日志行列表,列表第一个元素可能是一个完整的日志行,也可能是上一个数据块末尾日志行的组成部分,列表最后一个元素可能是不完整的日志行(即下一个数据块开头日志行的组成部分),也可能是空字符串(日志块中的日志行数据全部是完整的),根据这个规律,得出以下公式,通过该公式,可以得到一个新的数据块,对该数据块二次切分,可以得到数据完整的日志行

上一个日志块首部日志行 +\n + 尾部日志行 + 下一个数据块首部日志行 + \n + 尾部日志行 + ...

3、将数据解析操作拆分为可并行解析部分和不可并行解析部分

数据解析往往涉及一些不可并行的操作,比如数据求和,最值统计等,如果不进行拆分,并行解析时势必需要添加互斥锁,避免数据覆盖,这样就会大大降低执行的效率,特别是不可并行操作占比较大的情况下。

对数据解析操作进行拆分后,可并行解析操作部分不用加锁。考虑到Python GIL的问题,不可并行解析部分替换为单进程解析。

4、采用多进程解析替代多线程解析

采用多进程解析替代多线程解析,可以避开Python GIL全局解释锁带来的执行效率问题,从而提高解析效率。

5、采用队列实现“协同”效果

引入队列机制,实现一边读取日志,一边进行数据解析:

  1. 日志读取线程将日志块存储到队列,解析进程从队列获取已读取日志块,执行可并行解析操作
  2. 并行解析操作进程将解析后的结果存储到另一个队列,另一个解析进程从队列获取数据,执行不可并行解析操作。

代码实现

#!/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每次是一个比较高效的配置。

原文链接

扫描二维码推送至手机访问。

版权声明:本文来源于网络,仅供学习,如侵权请联系站长删除。

本文链接:https://news.layui.org.cn/post/1176.html

分享给朋友:

“Python 大数据量文本文件高效解析方案代码实现” 的相关文章

HBase1.4.6安装搭建及shell命令使用

HBase1.4.6安装搭建 目录 HBase1.4.6安装搭建 一、前期准备(Hadoop,zookeeper,jdk) 搭建Hbase 1、上传解压 2、配置环境变量 3、修改hbase-env.sh文件 4、修改hbase-site.xml文件 5、修改regionservers文件 6、同步到所有节点(如果是伪分布式不需要同步) 7、启动hbase集群 , 在master上执行...

一篇文章带你掌握主流服务层框架——SpringMVC

一篇文章带你掌握主流服务层框架——SpringMVC 在之前的文章中我们已经学习了Spring的基本内容,SpringMVC隶属于Spring的一部分内容 但由于SpringMVC完全针对于服务层使用,所以我们在介绍时常常把SpringMVC单独当作一个大章节来学习 温馨提醒:在学习SpringMVC前请确保已学习Spring内容 SpringMVC简介 首先我们先来简单了解一下SpringM...

SqlKata - 方便好用的 Sql query builder

SqlKata查询生成器是一个用C# 编写的功能强大的Sql查询生成器。它是安全的,与框架无关。灵感来源于可用的顶级查询生成器,如Laravel Query Builder和 Knex。 SqlKata有一个富有表现力的API。它遵循一个干净的命名约定,这与SQL语法非常相似。 通过在受支持的数据库引擎上提供一个抽象级别,允许您使用相同的统一API处理多个数据库,并且可搭配 dapper 使用,可...

IPv6报文头深度解析

IPv6报文由IPv6基本报文头、IPv6扩展报文头以及上层协议数据单元3部分组成。上层协议数据单元一般由上层协议报文头和它的有效载荷构成,上层协议数据单元可以是一个ICMPv6报文、一个TCP报文或一个UDP报文。 1、IPv6基本报文头 IPv6基本报文头有8个字段,固定大小为40 Byte,每一个IPv6数据报都必须包含基本报文头。基本报文头提供报文转发的基本信息,由转发路径上的所有设备解...

多功能手持VH501TC采集仪如何设置振弦传感器的激励方法和激励电压

河北稳控科技多功能手持VH501TC采集仪如何设置振弦传感器的激励方法和激励电压 VH501TC 提供多种振弦传感器激励方法,以最大限度兼容所有厂家和型号的振弦传感器。 振弦传感器激励方法参数位于实时数据窗口右侧,共有 5 种方法可选,分别用 MODTH0~MODTH4 表示。各方法说明如下: 激励电压数据在屏幕上显示为 xxx/xxx 的形式,其中前面的数字表示实际的激励电压,后面的数字...

打造企业自己代码规范IDEA插件(中)

一些基本概念 在开始独立研发公司自己的代码规范检查规则之前,先介绍一些相关的基本概念。阿里巴巴代码规范很多规则其实都是基于开源框架PMD进行的研发。PMD用官方的话语介绍来说:PMD是一个源代码分析器。它可以发现常见的编程缺陷,如未使用的变量、空catch块、不必要的对象创建等。它支持多种语言。它可以用自定义规则进行扩展。它使用JavaCC和Antlr将源文件解析为抽象语法树(AST),并对其运...

发表评论

访客

看不清,换一张

◎欢迎参与讨论,请在这里发表您的看法和观点。