Reading a large XML file and parsing necessary elements into MySQLdb

Question

I have fair concept in a programming (learner) but not an expert to refactor code at the highest level. I am trying to read a huge (100MB-2GB) XML file and parse necessary element (attributes) from a file into MySQLdb. The code is working perfectly as I have tested in small file sizes. But, unfortunately when I got big size today (400MB-900MB), it's taking unexpected time.

Currently, I am diving into List Comprehension, generator(), lambda() and list reduction techniques.

Not to make more lines of code, but I have eliminated 3 or 4 elements from my list (taglist) from here. But, the structure of the code for rest of the eliminated elements are the same.

I am using: Python 2.7, lxml, MySQL (1.2.3)

import os, sys
import stat
import getpass
import MySQLdb
from lxml import etree
import datetime, time
import dbconfig as config

# All global variables set
 #Namespace which is default in every mzML file
#taglist=['mzML','sourceFile','software','dataProcessing','instrumentConfiguration','selectedIon']
taglist=['mzML','sourceFile']
NS="{http://psi.hupo.org/ms/mzml}"

def fast_iter(context, func,args=[],kwargs={}):
    # fast_iter is useful if we need to free memory while iterating through a
    # very large XML file.
    #http://www.ibm.com/developerworks/xml/library/x-hiperfparse/
    # Author: Liza Daly

    for event, elem in context:
        func(elem,*args, **kwargs)
        elem.clear()
        while elem.getprevious() is not None:
            del elem.getparent()[0]
    del context

def process_element(elt):
    # global mzml_pk, sf_cv_fk,softw_fk_ID, softw_cv_fk_ID
    # Processing first element (mzML) and it's attributes (id, version, accession)
    # version attribute is required AND id & accession are optional
    if elt.tag==NS+taglist[0]:

        L= elt.keys()
        L1=['id','version','accession']

        # Checking whether element attributes (items L1) exist in a mzml element attributes (L)
        if L1[0] in L:
            mzmlID = elt.attrib['id']
        else:
            mzmlID = "-"
        if L1[1] in L:
            mzml_version = elt.attrib['version']
        else:
            mzml_version = "-"
        if L1[2] in L:
            mzml_accession = elt.attrib['accession']
        else:
            mzml_accession = "-"

        exp_id_fk = exp_PK

        sql = """INSERT INTO pmass_mzml (mzml_id,accession,version,exp_id)
        VALUES (%s, %s, %s, %s)"""
        try:
            # Execute the SQL command
            config.cursor.execute(sql,(mzmlID,mzml_accession,mzml_version,exp_id_fk))
            # Commit your changes in the database
            config.conn.commit()

            #mzml_pk= cursor.lastrowid
        except Exception as err:
            # logger.error(err)
            # Rollback in case there is any error
            config.conn.rollback()
        global mzml_pk
        mzml_pk = config.cursor.lastrowid
    # Processing second element (sourceFile) which has attributes (id, name and location)
    # SourceFile attributes id, name and location are required
    # Further, sourceFile has child element (cvParam) and attributes (cvRef, name, accession, value)
    # cvParam attributes: cvRef, name, accession are Required and value optional

    elif elt.tag==NS+taglist[1]:
        sf_keys = elt.keys()
        sf_need = ['id','name','location']
        if sf_need[0] in sf_keys:
            sf_id = elt.attrib['id']
        else:
            sf_id, "-"
        if sf_need[1] in sf_keys:
            sf_name = elt.attrib['name']
        else:
           sf_name, "-"
        if sf_need[2] in sf_keys:
            sf_location = elt.attrib['location']
        else:
            sf_location = "-"
        global sf_fk
        sf_fk = mzml_pk
        #print "Insert values into django sourceFile class data model"

        sql = """INSERT INTO pmass_source_file (sf_id,name,location,mzml_fk_id)
        VALUES (%s, %s, %s, %s)"""
        try:
            # Execute the SQL command
            config.cursor.execute(sql,(sf_id,sf_name,sf_location,sf_fk))
            # Commit your changes in the database
            config.conn.commit()

        except Exception as err:
            # logger.error(err)
            # Rollback in case there is any error
            config.conn.rollback()
        #sf_pk = cursor.lastrowid
        global sf_cv_fk
        sf_cv_fk= config.cursor.lastrowid

        for child in elt.getchildren():
            sf_child_keys =child.keys()
            sf_child_need = ['cvRef','name','accession','value']

            if sf_child_need[0] in sf_child_keys:
                sf_cv_cvref = child.attrib['cvRef']
            else:
                sf_cv_cvref = "-"
            if sf_child_need[1] in sf_child_keys:
                sf_cv_name = child.attrib['name']
            else:
                sf_cv_name = "-"
            if sf_child_need[2] in sf_child_keys:
                sf_cv_accession = child.attrib['accession']
            else:
                sf_cv_accession = "-"
            if sf_child_need[3] in sf_child_keys :
                if len(child.get('value'))>0:
                    sf_cv_value =  child.attrib['value']
                else:
                    sf_cv_value = "-"
            else:
                sf_cv_value = "-"

            sql = """INSERT INTO pmass_source_file_cv (ref, accession,name,value,sf_fk_id)
                    VALUES (%s, %s, %s, %s, %s)"""
            try:
                # Execute the SQL command
                config.cursor.execute(sql,(sf_cv_cvref,sf_cv_accession,sf_cv_name,sf_cv_value,sf_cv_fk))
                # Commit your changes in the database
                config.conn.commit()
            except Exception as err:
                # logger.error(err)
                # Rollback in case there is any error
                config.conn.rollback()

I haven't pasted the rest of the code for other elements because it's more than 300 lines, but of the same concept.

def convert_bytes(bytes):
    bytes = float(bytes)
    if bytes >= 1099511627776:
        terabytes = bytes / 1099511627776
        size = '%.2fTB' % terabytes
    elif bytes >= 1073741824:
        gigabytes = bytes / 1073741824
        size = '%.2fGB' % gigabytes
    elif bytes >= 1048576:
        megabytes = bytes / 1048576
        size = '%.2fMB' % megabytes
    elif bytes >= 1024:
        kilobytes = bytes / 1024
        size = '%.2fKB' % kilobytes
    else:
        size = '%.2fb' % bytes
    return size


def main():
    global EXP_PK
    # Start time counter
    begin=time.clock()

    # Reading file from the configured directory
    file_dir = r'D:\files\111102_CA2.mzML'

    # Reactor so it will read all new files from configured directory
    file_name = os.path.basename(file_dir)

    # File name and extension
    (name, ext) = os.path.splitext(file_name)
    user= getpass.getuser()
    exp_name = name
    exp_type= ext[1:]
    exp_created_time = time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(os.path.getctime(file_dir)))
    exp_modi_time = time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(os.path.getmtime(file_dir)))
    exp_uploaddate= datetime.date.today()
    exp_uptime = str(datetime.datetime.now())[10:19]
    exp_size = convert_bytes(os.path.getsize(file_dir))
    #exp_located = r'C:\Users\Thaman\Documents\My Dropbox\Files\small1.mzML'
    exp_located = r'C:\My Documents\Dropbox\Files\plgs_example.mzML'
    #exp_located =r'D:\files\111102_CA2.mzML'

    sql = """INSERT INTO pmass_experiment (user,filename,filetype,createddate,modifieddate,uploaddate,time,size,located)
         VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)"""
    try:
        # Execute the SQL command
        config.cursor.execute(sql,(user,exp_name,exp_type,exp_created_time,exp_modi_time,exp_uploaddate,exp_uptime,exp_size,exp_located))
        # Commit your changes in the database
        config.conn.commit()
    except:
        #logger.error(err)
        # Rollback in case there is any error
        config.conn.rollback()
    global exp_PK
    exp_PK = config.cursor.lastrowid
    #Reading elements from file
    for intag in range(len(taglist)):
        context= etree.iterparse(file_dir,events=("end",),tag=NS+taglist[intag])
        fast_iter(context, process_element)

    tend= time.clock()
    print "\n\nTIME TAKEN TO COMPLETE JOB", tend-begin

if __name__ == '__main__':
    main()

Problem according to me:

• I am sure my approach towards using too much if and for expression is a problem but haven't figure out the solution.
• 400 MB file is still and taking more than 1 hr.

Reviewing: Pythonic, memory friendly, speed optimization, structure

If someone wants the file, I can share it on Dropbox.

Answer 1

import os, sys
import stat
import getpass
import MySQLdb
from lxml import etree
import datetime, time
import dbconfig as config

They python style guide recommends one import per line. I'd also probably group things together i.e. standard library imports then 3rd party imports/ then app imports. But that doesn't really matter much

# All global variables set
 #Namespace which is default in every mzML file

taglist=['mzML','sourceFile']
NS="{http://psi.hupo.org/ms/mzml}"

For global constants, the python style guide recommends ALL_CAPS. I recommend not using NS, instead spell it out NAMESPACE

def fast_iter(context, func,args=[],kwargs={}):

Taking variable arguments as a list and dict are odd, any reason you didn't take them as *args and **kwargs?

    # fast_iter is useful if we need to free memory while iterating through a
    # very large XML file.
    #http://www.ibm.com/developerworks/xml/library/x-hiperfparse/
    # Author: Liza Daly

    for event, elem in context:
        func(elem,*args, **kwargs)

Using a yield to make this a generator rather then calling a function would probable be better.

        elem.clear()
        while elem.getprevious() is not None:
            del elem.getparent()[0]
    del context

There is no point in deleting local names at the end of the function. When the function ends all the names will be deleted anyways.

def process_element(elt):

I recommend not using short abbreviations like elt unless they are really common.

    # global mzml_pk, sf_cv_fk,softw_fk_ID, softw_cv_fk_ID
    # Processing first element (mzML) and it's attributes (id, version, accession)
    # version attribute is required AND id & accession are optional
    if elt.tag==NS+taglist[0]:

        L= elt.keys()
        L1=['id','version','accession']

        # Checking whether element attributes (items L1) exist in a mzml element attributes (L)
        if L1[0] in L:

You haven't helped anything by storing 'id' L1[0] and then fetching it again. Just use 'id' here.

            mzmlID = elt.attrib['id']
        else:
            mzmlID = "-"

You should be able to do something like mzmlID = el.attrib.get('id', '-') to replace this whole if/else block.

        if L1[1] in L:
            mzml_version = elt.attrib['version']
        else:
            mzml_version = "-"
        if L1[2] in L:
            mzml_accession = elt.attrib['accession']
        else:
            mzml_accession = "-"

        exp_id_fk = exp_PK

        sql = """INSERT INTO pmass_mzml (mzml_id,accession,version,exp_id)
        VALUES (%s, %s, %s, %s)"""
        try:
            # Execute the SQL command
            config.cursor.execute(sql,(mzmlID,mzml_accession,mzml_version,exp_id_fk))
            # Commit your changes in the database
            config.conn.commit()

            #mzml_pk= cursor.lastrowid

Don't leave dead code in comments

        except Exception as err:
            # logger.error(err)
            # Rollback in case there is any error
            config.conn.rollback()

Don't just ignore errors that happen. If you expect an error to occur, you should check to make sure exactly the error you wanted was caught. Otherwise you should re-raise or at least log the error.

You might also look into with statments and context managers. Amongst other things, they make transactions easier to work with.

        global mzml_pk
        mzml_pk = config.cursor.lastrowid

avoid use of global variables. Use return values/object attributes/pretty much anything before using a global variable.

    # Processing second element (sourceFile) which has attributes (id, name and location)
    # SourceFile attributes id, name and location are required
    # Further, sourceFile has child element (cvParam) and attributes (cvRef, name, accession, value)
    # cvParam attributes: cvRef, name, accession are Required and value optional

Right about now, this function is way too long. You should break it up into several functions.

    elif elt.tag==NS+taglist[1]:
        sf_keys = elt.keys()
        sf_need = ['id','name','location']
        if sf_need[0] in sf_keys:
            sf_id = elt.attrib['id']
        else:
            sf_id, "-"
        if sf_need[1] in sf_keys:
            sf_name = elt.attrib['name']
        else:
           sf_name, "-"
        if sf_need[2] in sf_keys:
            sf_location = elt.attrib['location']
        else:
            sf_location = "-"
        global sf_fk
        sf_fk = mzml_pk
        #print "Insert values into django sourceFile class data model"

        sql = """INSERT INTO pmass_source_file (sf_id,name,location,mzml_fk_id)
        VALUES (%s, %s, %s, %s)"""
        try:
            # Execute the SQL command
            config.cursor.execute(sql,(sf_id,sf_name,sf_location,sf_fk))
            # Commit your changes in the database
            config.conn.commit()

        except Exception as err:
            # logger.error(err)
            # Rollback in case there is any error
            config.conn.rollback()
        #sf_pk = cursor.lastrowid
        global sf_cv_fk
        sf_cv_fk= config.cursor.lastrowid

Deja vu... You should see if you can refactor the simiairites between any two very similiar pieces of code like this.

        for child in elt.getchildren():
            sf_child_keys =child.keys()
            sf_child_need = ['cvRef','name','accession','value']

            if sf_child_need[0] in sf_child_keys:
                sf_cv_cvref = child.attrib['cvRef']
            else:
                sf_cv_cvref = "-"
            if sf_child_need[1] in sf_child_keys:
                sf_cv_name = child.attrib['name']
            else:
                sf_cv_name = "-"
            if sf_child_need[2] in sf_child_keys:
                sf_cv_accession = child.attrib['accession']
            else:
                sf_cv_accession = "-"
            if sf_child_need[3] in sf_child_keys :
                if len(child.get('value'))>0:
                    sf_cv_value =  child.attrib['value']
                else:
                    sf_cv_value = "-"
            else:
                sf_cv_value = "-"

            sql = """INSERT INTO pmass_source_file_cv (ref, accession,name,value,sf_fk_id)
                    VALUES (%s, %s, %s, %s, %s)"""
            try:
                # Execute the SQL command
                config.cursor.execute(sql,(sf_cv_cvref,sf_cv_accession,sf_cv_name,sf_cv_value,sf_cv_fk))
                # Commit your changes in the database
                config.conn.commit()
            except Exception as err:
                # logger.error(err)
                # Rollback in case there is any error
                config.conn.rollback()

# **Rest of the codes I have pasted because it's more then 300 lines but same concept reading other element**

You've got lots of repeating sections which is obnoxious to deal with. I'd try something like. You should pull any common elements into a function or perhaps but all the different elements in a big list/dictionary and pull data from there

def convert_bytes(bytes):

This function doesn't really convert bytes, it formats the count of bytes. The name could be better.

    bytes = float(bytes)

Why are you converting bytes to a float? I presume you don't have fractional bits

    if bytes >= 1099511627776:
        terabytes = bytes / 1099511627776
        size = '%.2fTB' % terabytes
    elif bytes >= 1073741824:
        gigabytes = bytes / 1073741824
        size = '%.2fGB' % gigabytes
    elif bytes >= 1048576:
        megabytes = bytes / 1048576
        size = '%.2fMB' % megabytes
    elif bytes >= 1024:
        kilobytes = bytes / 1024
        size = '%.2fKB' % kilobytes
    else:
        size = '%.2fb' % bytes
    return size

The code repeats the same idea several times, I'd use a data-based approach

UNITS = [ 
    (1024**4, 'TB'),
    (1024**3, 'GB'),
    (1024**2, 'MB'),
    (1024**1, 'KB'),
    (1024**0, 'b')
]

def convert_bytes(bytes):
    for size, unit in UNITS:
        if bytes > size:
           return '%d%s' % (bytes / size, unit)



def main():
    global EXP_PK
    # Start time counter
    begin=time.clock()

    # Reading file from the configured directory
    file_dir = r'D:\files\111102_CA2.mzML'

That's not a directory...

    # Reactor so it will read all new files from configured directory
    file_name = os.path.basename(file_dir)

Reactor? Did you mean refactor? Because that's not what that means either. Really, I can't heads or tails out of your comment

    # File name and extension
    (name, ext) = os.path.splitext(file_name)
    user= getpass.getuser()
    exp_name = name
    exp_type= ext[1:]

A comment explaining why you are stripping off the first character of the extension might be nice.

    exp_created_time = time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(os.path.getctime(file_dir)))
    exp_modi_time = time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(os.path.getmtime(file_dir)))
    exp_uploaddate= datetime.date.today()
    exp_uptime = str(datetime.datetime.now())[10:19]
    exp_size = convert_bytes(os.path.getsize(file_dir))
    #exp_located = r'C:\Users\Thaman\Documents\My Dropbox\Files\small1.mzML'
    exp_located = r'C:\My Documents\Dropbox\Files\plgs_example.mzML'
    #exp_located =r'D:\files\111102_CA2.mzML'

    sql = """INSERT INTO pmass_experiment (user,filename,filetype,createddate,modifieddate,uploaddate,time,size,located)
         VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)"""
    try:
        # Execute the SQL command
        config.cursor.execute(sql,(user,exp_name,exp_type,exp_created_time,exp_modi_time,exp_uploaddate,exp_uptime,exp_size,exp_located))
        # Commit your changes in the database
        config.conn.commit()
    except:
        #logger.error(err)
        # Rollback in case there is any error
        config.conn.rollback()
    global exp_PK
    exp_PK = config.cursor.lastrowid

You have this basic piece of code over and over again. That's a very big hint you should move it into a function and reuse it.

    #Reading elements from file
    for intag in range(len(taglist)):

Just use for tag in taglist: You aren't using the index, so there is no reason to use range

        context= etree.iterparse(file_dir,events=("end",),tag=NS+taglist[intag])
        fast_iter(context, process_element)

    tend= time.clock()
    print "\n\nTIME TAKEN TO COMPLETE JOB", tend-begin

if __name__ == '__main__':
    main()

EDIT

You should do something like this in order to handle the different tag types. It'll be faster and clearer then what you are doing.

def process_foo_tag(element):
    fetch element from foo
    insert into database

def process_bar_tag(element):
    fetch elements for bar tag
    insert into database

TAGS = {
   'bar' : process_bar_tag,
   'foo' : process_foo_tag
}

def process_element(element):
    TAGS[element.tag](element)