一、前言

最近在做离线数据导入HBase项目，涉及将存储在Mysql中的历史数据通过bulkload的方式导入HBase。由于源数据已经不在DB中，而是以文件形式存储在机器磁盘，此文件是mysqldump导出的格式。如何将mysqldump格式的文件转换成实际的数据文件提供给bulkload作转换，是需要考虑的一个问题。

二、思路

我们知道mysqldump导出的文件主要是Insert，数据库表结构定义语句。而要解析的对象也主要是包含INSERT关键字记录，这样我们就把问题转换成如何从dmp文件解析Insert语句。接触过dmp文件的同学应该了解，其INSERT语句的结构，主要包含表名、字段名、字段值, 这里面主要包含几个关键字：INSERT INTO, VALUES。我们要做的就是把Values括号后的字段值给解析出来，这个过程需要考虑VALUES后面包含的是多少行的记录，有可能导出的记录Values后面包含多行对应mysql中存储的记录。

在解析文件过程中，我自然想到用Python来写，因为Python在处理文件方面有很多优势，也比较简单。在处理DMP文件这块，考虑到字段值间是用逗号分割的，在python中正好一个模块可以很好的来处理此类格式，即大家很熟悉的CSV模块，在处理CSV类型的文件有很多优势。在这里我们把CSV模块有在解析dmp文件，同时加一些解析逻辑，可以很好解决此类问题。

同时，我们要处理的dmp文件是经过压缩的，并且单个文件都比较大，都是Gigbytes的，在读取时需要注意机器内存大小，不能一次读出所有的数据，python也考虑到此类问题，采用的方法是惰性取值，即在真正使用时才从磁盘中加载相应的文件数据。如果想加块解析，还可以采集多进程或多线程的方法。

三、方法

处理流程图如下所示：

代码如下图所示：

 #!/usr/bin/env python

 import fileinput

 import csv

 import sys

 import gzip

 # 设定CSV读取的最大容量

 csv.field_size_limit(sys.maxsize)

 def check_insert(line):

     """

     返回语句是否以insert into开头，如果是返回true,否则返回false

     """

     return line.startswith('INSERT INTO') or False

 def get_line_values(line):

     """

     返回Insert语句中包含Values的部分

     """

     return line.partition('VALUES ')[2]

 def check_values_style(values):

     """

     保证INSERT语句满足基本的条件，即包含(右括号

     """

     if values and values[0] == '(':

         return True

     return False

 def parse_line(values):

     """

     创建csv对象，读取INSERT VALUES 字段值

     """

     latest_row = []

     reader = csv.reader([values], delimiter=',',

                         doublequote=False,

                         escapechar='\\',

                         quotechar="'",

                         strict=True

     )

     for reader_row in reader:

         for column in reader_row:

             # 判断字段值是否为空或为NULL

             if len(column) == 0 or column == 'NULL':

                 latest_row.append("")

                 continue

             # 判断字段开头是否以（开头,如果是则说明此VALUES后面不只包含一行数据，可能有多行，要分别解析

             if column[0] == "(":

                 new_row = False

                 if len(latest_row) > 0:

                     #判断行是否包含）,如果包含则说明一行数据完毕

                     if latest_row[-1][-1] == ")":

                         # 移除）

                         latest_row[-1] = latest_row[-1][:-1]

                         if latest_row[-1] == "NULL":

                             latest_row[-1] = ""

                         new_row = True

                 # 如果是新行，则打印该行

                 if new_row:

                     line="}}}{{{".join(latest_row)

                     print "%s<{||}>" % line

                     latest_row = []

                 if len(latest_row) == 0:

                     column = column[1:]

             latest_row.append(column)

         # 判断行结束符

         if latest_row[-1][-2:] == ");":

             latest_row[-1] = latest_row[-1][:-2]

             if latest_row[-1] == "NULL":

                 latest_row[-1] = ""

             line="}}}{{{".join(latest_row)

             print "%s<{||}>" % line

 def main():   

     filename=sys.argv[1]

     try:

         #惰性取行

         with gzip.open(filename,"rb") as f:

             for line in f:

                     if check_insert(line):

                         values = get_line_values(line)

                         if check_values_style(values):

                             parse_line(values)

     except KeyboardInterrupt:

         sys.exit(0)

 if __name__ == "__main__":

     main()

四、总结

总的说来，主要是利用Python的CSV模块来解析DMP文件的INSERT语句，如果DMP文件不规整，可能还是有些问题。对于dmp文件很大情况，也是需要考虑解析时间效率问题，可以考虑增加多进程或多线程机制。

巴特西

如何用python解析mysqldump文件

一、前言

二、思路

三、方法

四、总结

最新文章

热门文章