2
\$\begingroup\$

I have the code below, which works successfully, and is used to parse, clean log files (very large in size) and output into smaller sized files. This would take about 12-14 mins to process 1 GB worth of logs (on my laptop). Can this be made faster? Could Dask or parallelism or asyncio or another help speed this up?

I am new to python and pandas, and I have googled around, but am totally confused and cant seem to adopt any of the examples I saw.

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 log file

"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 Output

$ 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 errorfile
wrongemailfoo.com
\$\endgroup\$
1
\$\begingroup\$

I think there is quite a lot that could be improved on in your approach. My main piece of advice is to try and process each line in the data only once, since each line is independent you should be able to do this.

I'm not too familiar with pandas but it seems like there are two main areas of concern.

  1. The section where you clean up the data and filter out all the bad emails, you create a mask by executing two regexs on each line and then read through and make copies of the data frame twice using the mask. At this point you have passed over every line in the data 3 times.
    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. The second section where you breakdown each email into a different file if it is valid. you go through every line in the dataframe for every possible starting letter, and copy the result over to process again. At this point you have gone through each line in the data about 40 times.
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')

Running cProfile on the code, when it just has to read one file with 6 lines in it produces this: 336691 function calls (328148 primitive calls) in 0.974 seconds. Nearly a second to just read and process 6 lines into different files is not good.

Rather than taking a pandas approach I have just written a pure python script that sketches out an alternative strategy. Doing the same test with cProfile produces 11228 function calls (11045 primitive calls) in 0.038 seconds. It might not fit your needs exactly but you could look at it for ideas about how to tweak your script.

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()

```
| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Thanks, interesting idea. And agreed that pandas is not fit for purpose. However, your code did not speed things up too much. Can this code be run in parallel ? async or similar? \$\endgroup\$ – rogerwhite Jun 18 at 5:59
  • \$\begingroup\$ It would be possible to adapt it in that way, you would need to make sure that different threads don't start reading from the same files and that different threads can output to the same log file in a safe way. \$\endgroup\$ – MindOfMetalAndWheels Jun 18 at 13:21

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.