最近在做离线数据导入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也考虑到此类问题,采用的方法是惰性取值,即在真正使用时才从磁盘中加载相应的文件数据。如果想加块解析,还可以采集多进程或多线程的方法。

二、方法

处理流程图如下所示:

解析流程图

代码如下所示:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
#!/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文件很大情况,也是需要考虑解析时间效率问题,可以考虑增加多进程或多线程机制。

版权声明:本文为博主原创文章,未经博主允许不得转载。