数据处理

2017-03-22

前言

最近在处理各种数据
数据源一般有普通文本、Access、csv、MSSQL等，比较杂乱
而针对不同格式的处境，又要编写不同的脚本，将数据整理之后放入到MongoDB中
记录如下：

文本格式

场景: 存在相同格式的多个文本数据，文件结构如下

代码实现

由于数据不便于公开，所以之贴出部分关键代码
具体思路是从文件中读取数据后拼接成dict，然后append到list里
最后在通过pymongo的insert_many将list插入到MongoDB中

# -*- coding: utf-8 -*-
from IPython import embed
from pymongo import MongoClient
client = MongoClient(connect=False)
db = client.info
import sys
reload(sys)
sys.setdefaultencoding('utf-8')
FILE_LIST = ['1.txt', '2.txt', '3.txt', '4.txt']

SQL Server

对于SQL Server的数据源，一般是mdf文件；直接附加到SQL Server 数据库中然后通过Python连接到数据库(使用pyodbc)，执行SQL语句……

代码实现

部分敏感字段已删除

# -*- coding: utf-8 -*-
import pyodbc
from IPython import embed
from pymongo import MongoClient
import sys
reload(sys)
sys.setdefaultencoding('utf-8')
client = MongoClient(connect=False)
db = client.cninfo
server = '192.168.1.143'
database = 'auction'
username = 'sa'
password = 'pass'
driver= '{ODBC Driver 13 for SQL Server}'
CONNECT_STR = 'DRIVER='+driver+';PORT=1433;SERVER='+server+';PORT=1443;DATABASE='+database+';UID='+username+';PWD='+ password
SELECT_DATA_STR = ("SELECT *****;")
def get_data_from_datas():
    try:
        cursor.execute(SELECT_DATA_STR)
        return cursor.fetchall()
    except:
        print 'error'
def data2json(data):
    """"将数据格式转换为json"""
    post = data[4] if data[4] else ''
    addr = u'{}{}'.format(data[5], data[6])
    office_tel = data[8] if data[8] else ''
    mobile_tel = data[9] if data[9] else ''
    url = data[10] if data[10] else ''
    f_email1 = data[11] if data[11] else ''
    f_email2 = data[12] if data[12] else ''
    item = {
        "tips": tips_str
    }
    return item
def insert_to_mongo(json_list):
    """将数据插入mongo"""
    db.sedb.insert_many(json_list)
    pass
if __name__ == '__main__':
    cnxn = pyodbc.connect(CONNECT_STR)
    cursor = cnxn.cursor()
    json_list = []
    datas = get_data_from_datas()
    for i in datas:
        json_list.append(data2json(i))
        if len(json_list) == 2000:
            print "2000 line data inserting……"
            insert_to_mongo(json_list)
            json_list = []
    if json_list:
        print "inserting {} line".format(len(json_list))
        insert_to_mongo(json_list)
        json_list = []

补充

又有一个场景是像弄两个MongoDB的数据整合
于是就用Python编写了个多线程脚本，不断充源数据库中读取数据后存放到目标数据库中，代码如下

# -*- coding: utf-8 -*-
from pymongo import MongoClient
from IPython import embed
import argparse
import threading
import Queue
import sys
reload(sys)
sys.setdefaultencoding('utf8')
parser = argparse.ArgumentParser(description='data migration')
parser.add_argument('thread', type=int, help='write thread count')
args = parser.parse_args()
queue = Queue.Queue()
host = '192.168.1.11'
new_port = 27018
client = MongoClient(host=host)
client15t = MongoClient(host=host, port=new_port)
cninfo = client.cninfo
sedb = cninfo.sedb
social_db = client15t.social_engine
hk = social_db.hongkong
def run():
    while hk.count():
        result = {}
        data = hk.find_one()
        result['realname'] = data['name']
        result['username'] = data['username']
        result['email'] = data['email']
        result['password'] = data['password']
        result['qq'] = data['phone']
        result['shenfenzheng'] = data['id_card']
        result['site'] = 'hk1t'
        result['tips'] = u'ip:{},addr:{},source:{}'.format(
            data['ip'], data['address'], data['source'])
        try:
            if hk.find_one_and_delete({'_id': data['_id']}):
                sedb.insert(result)
                print hk.count()
        except:
            print 'Data has been deleted. '
            pass
        # print data
threads = []
for i in range(args.thread):
    t = threading.Thread(target=run)
    t.start()
for t in threads:
    t.join()%

如果需要结合screen的话，可以通过screen -d -m 将相关任务放到screen中