python通过dom和sax方式解析xml的应用实例分享

xml.dom

需求
有一个表,里面数据量比较大,每天一更新,其字段可以通过xml配置文件进行配置,即,可能每次建表的字段不一样。

上游跑时会根据配置从源文件中提取,到入库这一步需要根据配置进行建表。

解决
写了一个简单的xml,配置需要字段及类型

上游读取到对应的数据

入库这一步,先把原表删除,根据配置建新表

xml文件

id

query
varchar(200)
false
query

pv
integer
false
pv

avg_money
integer
false

处理脚本

#!/usr/bin/python
# -*- coding:utf-8 -*-
#author: wklken
#desc: use to read db xml config.
#———————–
#2012-02-18 created
#———————-
import sys,os
from xml.dom import minidom, node
def read_dbconfig_xml(xml_file_path):
content = {}
root = minidom.parse(xml_file_path)
table = root.getelementsbytagname(“table”)[0]
#read dbname and table name.
table_name = table.getattribute(“name”)
db_name = table.getattribute(“db_name”)
if len(table_name) > 0 and len(db_name) > 0:
db_sql = “create database if not exists `” + db_name +”`; use ” + db_name + “;”
table_drop_sql = “drop ” + table_name + ” if exists ” + table_name + “;”
content.update({“db_sql” : db_sql})
content.update({“table_sql” : table_drop_sql })
else:
print “error:attribute is not define well! db_name=” + db_name + ” ;table_name=” + table_name
sys.exit(1)
#print table_name, db_name
table_create_sql = “create table ” + table_name +”(”
#read primary cell
primary_key = table.getelementsbytagname(“primary_key”)[0]
primary_key_name = primary_key.getelementsbytagname(“name”)[0].childnodes[0].nodevalue
table_create_sql += primary_key_name + ” integer not null auto_increment primary key,”
#print primary_key.toxml()
#read ordernary field
fields = table.getelementsbytagname(“field”)
f_index = 0
for field in fields:
f_index += 1
name = field.getelementsbytagname(“name”)[0].childnodes[0].nodevalue
type = field.getelementsbytagname(“type”)[0].childnodes[0].nodevalue
table_create_sql += name + ” ” + type
if f_index != len(fields):
table_create_sql += “,”
is_index = field.getelementsbytagname(“is_index”)[0].childnodes[0].nodevalue
table_create_sql += “);”
content.update({“table_create_sql” : table_create_sql})
#character set latin1 collate latin1_danish_ci;
print content
if __name__ == “__main__”:
read_dbconfig_xml(sys.argv[1])

涉及方法
root = minidom.parse(xml_file_path) 获取dom对象

root.getelementsbytagname(“table”) 根据tag获取节点列表

table.getattribute(“name”) 获取属性

primary_key.getelementsbytagname(“name”)[0].childnodes[0].nodevalue 获取子节点的值(id 得到id)

sax
需求
读取xml数据文件,文件较大,需要实时处理插入到数据库

xml文档

100000

北京,海淀区
437
1989
333
242
null
大学

1
2月14日

处理
sax处理时并不会像dom一样可以以类似节点的维度进行读取,它只有 开始标签 内容 结束标签 之分

处理思想是:通过一个handler,对开始标签,内容,结束标签各有一个处理函数

代码及注解
person 处理类

from xml.sax import handler,parsestring
class personhandler(handler.contenthandler):
def __init__(self, db_ops):
#db op obj
self.db_ops = db_ops
#存储一个person的map
self.person = {}
#当前的tag
self.current_tag = “”
#是否是tag之间的内容 ,目的拿到tag间内容,不受空白的干扰
self.in_quote = 0
#开始,清空map
def startelement(self, name, attr):
#以person,清空map
if name == “person”:
self.person = {}
#记录 状态
self.current_tag = name
self.in_quote = 1
#结束,插入数据库
def endelement(self, name):
#以person结尾 代表读取一个person的信息结束
if name == “person”:
#do something
in_fields = tuple([ (‘”‘ + self.person.get(i,””) + ‘”‘) for i in fields ])
print in_sql % in_fields
db_ops.insert( in_sql%(in_fields))
#处理
self.in_quote = 0
def characters(self, content):
#若是在tag之间的内容,更新到map中
if self.in_quote:
self.person.update({self.current_tag: content})

加上入库的完整代码

#!/usr/bin/python
# -*- coding:utf-8 -*-
#parse_person.py
#version : 0.1
#author : wukunliang@163.com
#desc : parse person.xml and out sql
import sys,os
import mysqldb
reload(sys)
sys.setdefaultencoding(‘utf-8’)
in_sql = “insert into person(id,sex,address,fansnum,summary,wbnum,gznum,blog,edu,work,renzh,brithday) values(%s, %s, %s, %s, %s, %s,
%s, %s, %s, %s, %s, %s)”
fields = (“id”,”sex”,”address”,”fansnum”,”summary”,”wbnum”,”gznum”,”blog”,”edu”,”work”,”renzh”,”brithday”)
#数据库方法
class db_connect:
def __init__(self, db_host, user, pwd, db_name, charset=”utf8″, use_unicode = true):
print “init begin”
print db_host, user, pwd, db_name, charset , use_unicode
self.conn = mysqldb.connection(db_host, user, pwd, db_name, charset=charset , use_unicode=use_unicode)
print “init end”
def insert(self, sql):
try:
n = self.conn.cursor().execute(sql)
return n
except mysqldb.warning, e:
print “error: execute sql ‘”,sql,”‘ failed”
def close(self):
self.conn.close()
#person 处理类
from xml.sax import handler,parsestring
class personhandler(handler.contenthandler):
def __init__(self, db_ops):
#db op obj
self.db_ops = db_ops
#存储一个person的map
self.person = {}
#当前的tag
self.current_tag = “”
#是否是tag之间的内容
self.in_quote = 0
#开始,清空map
def startelement(self, name, attr):
#以person,清空map
if name == “person”:
self.person = {}
#记录 状态
self.current_tag = name
self.in_quote = 1
#结束,插入数据库
def endelement(self, name):
#以person结尾 代表读取一个person的信息结束
if name == “person”:
#do something
in_fields = tuple([ (‘”‘ + self.person.get(i,””) + ‘”‘) for i in fields ])
print in_sql % in_fields
db_ops.insert( in_sql%(in_fields))
#处理
self.in_quote = 0
def characters(self, content):
#若是在tag之间的内容,更新到map中
if self.in_quote:
self.person.update({self.current_tag: content})
if __name__ == “__main__”:
f = open(“./person.xml”)
#如果源文件gbk 转码 若是utf-8,去掉decode.encode
db_ops = db_connect(“127.0.0.1”, “root”, “root”, “test”)
parsestring(f.read().decode(“gbk”).encode(“utf-8”), personhandler(db_ops))
f.close()
db_ops.close()

平时拿python来分析数据,工具脚本还有hadoop streamming,但是用的面和深度实在欠缺 只能说道行还浅,需要多多实践

Posted in 未分类