解析和清理日志文件

Question

下面的代码工作成功，用于解析、清理日志文件(非常大的大小)并输出到较小的文件中。这需要12到14分钟来处理1GB的日志(在我的笔记本电脑上)。这能更快些吗？Dask、并行、异步或其他方式能帮助加快速度吗？

我刚接触过蟒蛇和熊猫，我在谷歌上搜索过，但我完全困惑，似乎不能采纳我看到的任何例子。

import os
import pandas as pd

asciireg = "[^\x00-\x7F]+"
emailreg = "^\w+(?:[-+.']\w+)*@\w+(?:[-.]\w+)*\.\w+(?:[-.]\w+)*$"
    
for root, dirs, files in os.walk('.', topdown=True):
    for file in files:
        try:
            for df in pd.read_csv(file, sep='\n', header=None, engine='python', quoting=3, chunksize=1200000):
                df = df[0].str.strip(' \t"').str.split('[,|;: \t]+', 1, expand=True).rename(columns={0: 'email', 1: 'data'}) 
            
                mask = (df.email.str.contains(emailreg, regex=True, na=False)) & (~df.data.str.contains(asciireg, regex=True, na=False))
                df2 = df[~mask].copy()
                df = df[mask].copy()
                df2[['email', 'data']].to_csv("errorfile", sep=':', index=False, header=False, mode='a', compression='gzip')
                del df2
                del mask
        
            for x in "abcdefghijklmnopqrstuvwxyz0123456789":
                df2 = df[df.email.str.startswith(x)]
                if (df.email.size > 0):
                    df2[['email', 'data']].to_csv(x, sep=':', index=False, header=False, mode='a')
        except Exception as e: 
            print ("Error: ", file)
            print(str(e))
        else:
            os.remove(file)

Sample日志文件

"email1@foo.com:datahere2     
email2@foo.com:datahere2
email3@foo.com datahere2
email5@foo.com;dtat'ah'ere2 
wrongemailfoo.com
email3@foo.com:datahere2

Expected输出

$ cat e

email1@foo.com:datahere2     
email2@foo.com:datahere2
email3@foo.com:datahere2
email5@foo.com:dtat'ah'ere2 
email3@foo.com:datahere2

$ cat错误文件

wrongemailfoo.com

Answer 1

我认为在你的方法上还有很多可以改进的地方。我的主要建议是尝试并处理数据中的每一行一次，因为每一行都是独立的，因此您应该能够做到这一点。

我对熊猫不太熟悉，但似乎有两个主要方面值得关注。

1. 在清除数据并过滤掉所有坏邮件的部分，您通过在每行上执行两个regexs创建一个掩码，然后使用该掩码读取并复制两次数据帧。此时，您已经遍历了数据中的每一行3次。

    df = df[0].str.strip(' \t"').str.split('[,|;: \t]+', 1, expand=True).rename(columns={0: 'email', 1: 'data'}) 
    mask = (df.email.str.contains(emailreg, regex=True, na=False)) & (~df.data.str.contains(asciireg, regex=True, na=False))
    df2 = df[~mask].copy()
    df = df[mask].copy()
    df2[['email', 'data']].to_csv("errorfile", sep=':', index=False, header=False, mode='a', compression='gzip')
    del df2
    del mask

1. 第二部分，如果电子邮件是有效的，你可以把它分解成一个不同的文件。对每个可能的起始字母遍历dataframe中的每一行，然后将结果复制到处理中。此时，您已经遍历了数据中的每一行约40次。

for x in "abcdefghijklmnopqrstuvwxyz0123456789":
    df2 = df[df.email.str.startswith(x)]
    if (df.email.size > 0):
        df2[['email', 'data']].to_csv(x, sep=':', index=False, header=False, mode='a')

在代码上运行cProfile，当它只需要读取一个包含6行代码的文件时，就会产生这样的结果：336691 function calls (328148 primitive calls) in 0.974 seconds。将6行代码读取并处理成不同的文件几乎有一秒钟的时间是不好的。

我没有采用熊猫的方法，而是编写了一个纯python脚本，它勾勒出了另一种策略。使用cProfile进行同样的测试会产生11228 function calls (11045 primitive calls) in 0.038 seconds。它可能不完全适合您的需要，但您可以查看它的想法，如何调整您的脚本。

import re
import logging

EMAIL_REGEX = r"^\w+(?:[-+.']\w+)*@\w+(?:[-.]\w+)*\.\w+(?:[-.]\w+)*$"
OUTPUT_FILES = "abcdefghijklmnopqrstuvwxyz0123456789"


def configure_logging():
    """
    Configure a logger for each possible email start. 
    """

    # TODO - Tweak the handlers, output formats and locations 
    # to suit your needs

    error_handler = logging.FileHandler("error.log", mode="a")
    error_handler.setLevel(logging.ERROR)
    error_handler.setFormatter(logging.Formatter('%(message)s'))

    for entry in OUTPUT_FILES:
        logger = logging.getLogger(entry)
        handler = logging.FileHandler(f"{entry}.log", mode="a")
        handler.setFormatter(logging.Formatter('%(message)s'))
        handler.setLevel(logging.INFO)
        logger.addHandler(handler)
        logger.addHandler(error_handler)
        logger.setLevel(logging.INFO)
    
def gather_files():
    """
    Return all the log files that need to be processed.
    """
    # TODO - replace with your own logic to find files.
    return ["test_input.csv"]

def process_log_file(log_file_path):
    """
    For each line in the log file, process it once.
    """
    with open(log_file_path, "r") as log_file:
        for line in log_file:
            process_line(line)
                
def process_line(line):
    """
    Find the email and user from a line, test if the email is valid. Log the data
    to the appropriate place.
    """

    # TODO you may wish to change to logic 
    # to decide if the line is valid or not.

    line = line.strip(' \t"\n')
    data = re.split(r'[,|;: \t]+', line, maxsplit=1)
    logger = logging.getLogger(data[0][0])
    if len(data) == 2 and re.match(EMAIL_REGEX, data[0]):
        logger.info(":".join(data))
    else:
        logger.error(line)

def main():
    """
    Processes each log file in turn.
    """
    for log_file_path in gather_files():
        process_log_file(log_file_path)

if __name__ == "__main__":
    configure_logging()
    main()

```#qcStackCode#