使用python进行数据迁移案例

以你之姓@ 2022-07-14 15:32 211阅读 0赞

需要使用到mysql处理工具类:http://blog.csdn.net/u012572955/article/details/53666440

# coding=utf-8
    
    
    import sys
    from common import *
    
    
    def gen_db_rpt():
       reload(sys)
       sys.setdefaultencoding('utf8')
       #连接报表库
       rpt_db = connect_with(db_bi_conf)
       #连接小贷库
       base_db = connect_with(db_conf)
       # 清空目标表
       truncate_sql = 'truncate %s.%s' % (db_bi_conf['db'], 'tb_costreduction_detail')
       rpt_db.query(truncate_sql)
       # 提交事务
       rpt_db.commit()
    
       i = 0
       #一次查询的最大查询行数
       max_select_row = 100000
       while i >= 0:
          sql = """select d.SITENO SITENO,e.CUSTNAME CUSTNAME,a.applydate APPLYDATE,c.CONTRACTNO CONTRACTNO,
                    a.reasontype REASONTYPE, i.is_maiche ISSELLCARS,i.is_susong ISLAWSUIT,b.FEESTYPENAME FEESTYPENAME,
                    (case when b.FEESTYPENAME = '逾期利息' then SHOULDDELAYINT
                    when b.FEESTYPENAME = '上门催收费' then SHOULDSMCSF
                    when b.FEESTYPENAME = '违约金' then SHOULDDELAYAMT
                    when b.FEESTYPENAME = '利息' then SHOULDINT
                    when b.FEESTYPENAME = '管理费' then SHOULDMANAGE
                    when b.FEESTYPENAME = '拖车费' then SHOULDTCHF
                    when b.FEESTYPENAME = '停车费' then SHOULDTCF
                    when b.FEESTYPENAME = 'GPS费' then SHOULDGPSF
                    when b.FEESTYPENAME = '赎车费' then SHOULDSCF
                    when b.FEESTYPENAME = '风险金' then SHOULDFXJ
                    when b.FEESTYPENAME = '逾期违约金' then SHOULDPENATY
                    else 0 end) SHOULDRECEIVE,
                    (case when b.FEESTYPENAME = '逾期利息' then ACTDELAYINT
                    when b.FEESTYPENAME = '上门催收费' then ACTSMCSF
                    when b.FEESTYPENAME = '违约金' then ACTDELAYAMT
                    when b.FEESTYPENAME = '利息' then ACTINT
                    when b.FEESTYPENAME = '管理费' then ACTMANAGE
                    when b.FEESTYPENAME = '拖车费' then ACTTCHF
                    when b.FEESTYPENAME = '停车费' then ACTTCF
                    when b.FEESTYPENAME = 'GPS费' then ACTGPSF
                    when b.FEESTYPENAME = '赎车费' then ACTSCF
                    when b.FEESTYPENAME = '风险金' then ACTFXJ
                    when b.FEESTYPENAME = '逾期违约金' then ACTPENATY
                    else 0 end) ACTUALRECEIVE,
                    b.amt REDUCTION,b.phases phases,a.APPLYBY APPLYBY,
                    f.SUBMITUSER FINALJUDGEMENT,f.AUDITTIME FINALJUDGEDATE,if(d.DIRECTSELLER is not null,1,0) ISDIRECT,
                    a.status STATUS,a.remark REMARK
                    from tb_lf_specialfeesfree a
                    left join tb_lf_specialfeesfreesub b
                    on a.id = b.PID
                    left join (select * from tb_lm_payment group by id) c
                    on a.LOANID = c.id
                    left join tb_lb_applyinfo d
                    on c.APPLYID = d.id
                    left join tb_lb_custinfo e
                    on c.APPLYID = e.APPLYID
                    left join (select loanid,PHASES, SHOULDDELAYINT,
                    SHOULDSMCSF,SHOULDDELAYAMT,SHOULDINT,
                    SHOULDMANAGE,SHOULDTCHF,SHOULDTCF,
                    SHOULDGPSF,SHOULDSCF,SHOULDFXJ,SHOULDPENATY,
                    sum(ACTDELAYINT) ACTDELAYINT, sum(ACTSMCSF) ACTSMCSF, sum(ACTDELAYAMT) ACTDELAYAMT,
                    sum(ACTINT) ACTINT, sum(ACTMANAGE) ACTMANAGE,sum(ACTTCHF) ACTTCHF,sum(ACTTCF) ACTTCF,sum(ACTGPSF) ACTGPSF,
                    sum(ACTSCF) ACTSCF,sum(ACTFXJ) ACTFXJ,sum(ACTPENATY) ACTPENATY from tb_lf_returndetail group by loanid,PHASES
                    ) h
                    on h.LOANID = a.LOANID  and h.phases = a.phases
                    left join (select * from (select pid,SUBMITUSER,AUDITTIME from tb_lf_specialaudit where SPECIALTYPE = 5 ORDER BY id desc) a group by pid) f
                    on a.id = f.PID
                    left join (select PAYMENTID,if(situation =2,1,0) is_maiche,if(situation=6,1,0) is_susong from tb_la_urge group by PAYMENTID) i
                    on a.LOANID = i.PAYMENTID
                    where a.status in (3,4) and b.amt > 0 and b.FEESTYPENAME in ('逾期利息','上门催收费','违约金','利息','管理费','拖车费','停车费','GPS费','赎车费','风险金','逾期违约金')
                    LIMIT %d,%d""" %(i * max_select_row, max_select_row)
          result_r = base_db.querySql(sql,field=['SITENO', 'CUSTNAME', 'APPLYDATE', 'CONTRACTNO', 'REASONTYPE', 'ISSELLCARS',
                                            'ISLAWSUIT', 'FEESTYPENAME', 'SHOULDRECEIVE', 'ACTUALRECEIVE', 'REDUCTION',
                                            'PHASES', 'APPLYBY', 'FINALJUDGEMENT', 'FINALJUDGEDATE', 'ISDIRECT','STATUS', 'REMARK'])
          #循环条件设置
          if len(result_r) <= 0 and len(result_r) < max_select_row:
             i = -1
          else:
             i = i + 1
    
          inserted_num = 0
          #一次提交1万条数据 commit_num在common.py中被设置成了10000
          for value in result_r:
             try:
                rpt_db.insert('tb_costreduction_detail',value,commit=False)
                inserted_num += 1
                if inserted_num >= commit_num:
                   print '--------will commit[%d]' % inserted_num
                   rpt_db.commit()
                   inserted_num = 0
             except Exception,ex:
                print Exception,":",ex
          rpt_db.commit()
          if inserted_num > 0 and inserted_num < 10000:
             print '--------will commit[%d]' % inserted_num
    
    if __name__ == '__main__':
       start_time = datetime.datetime.today()
       print "report_ start time: %s" % (start_time)
       gen_db_rpt()
       end_time = datetime.datetime.today()
       print "report_ finish time: %s,inteval: %s" % (end_time, end_time - start_time)

连接信息:

db_rpt_conf = {}
    db_rpt_conf['host'] = '？'
    db_rpt_conf['db'] = '？'
    db_rpt_conf['user'] = 'root'
    db_rpt_conf['pwd'] = '123456'
    db_rpt_conf['port'] = 3306
    
    db_bi_conf = {}
    db_bi_conf['host'] = '？'
    db_bi_conf['db'] = '？'
    db_bi_conf['user'] = 'root'
    db_bi_conf['pwd'] = '123456'
    db_bi_conf['port'] = 3306