利用Python将EXCEL，CSV，TSV导入Oracle或者MySQL

Posted 2022-04-09 robinson1988

虽然可以用Kettle等ETL工具将EXCEL，CSV，TSV等格式的文件导入Oracle或者mysql，但是还是觉得写点代码心里踏实些

import cx_Oracle
import pymysql
import pandas
import os
import time
os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'
target_db_type='Oracle'                                                     #如果想导入MySQL，修改这里为MySQL即可
if target_db_type.upper()=='ORACLE':
  conn=cx_Oracle.connect('scott/tiger@192.168.3.13/orcl')                  #改成你自己的连接串
  cur =conn.cursor()
  cur.execute("alter session set nls_date_format='yyyy-mm-dd hh24:mi:ss'") #在导入数据库之前，先检查一下Date类型数据格式
elif target_db_type.upper()=='MYSQL':
  conn=pymysql.connect("192.168.56.101","scott","tiger","db")              #改成你自己的连接串
  cur = conn.cursor()
cur.arraysize = 1000
target_table=input("请输入要导入的表名:")
ifreadheader=input("是否包含表头(y,n):")                                     #CSV文件是否包含表头
if ifreadheader=='y':
 headervalue=0
elif ifreadheader=='n':
 headervalue=None
cur.execute('select * from '+target_table+' where 1=0')                    #获取列信息
val_str = ''
if target_db_type.upper()=='MYSQL':
 for i in range (1,len(cur.description)):
  val_str=val_str+'%s'+','
 val_str=val_str+'%s'                                                      #MySQL批量插入语法是 insert into tb_name values(%s,%s,%s,%s)
elif target_db_type.upper()=='ORACLE':
 for i in range (1,len(cur.description)):
  val_str=val_str+':'+str(i)+','
 val_str=val_str+':'+str(i+1)                                              #Oracle批量插入语法是 insert into tb_name values(:1,:2,:3)
insert_sql='insert into '+target_table+' values('+val_str+')'              #拼接insert into 目标表 values
print('开始导入:',time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
csv_data=pandas.read_csv('csvfile.txt',header=headervalue,chunksize=50000,encoding='GBK',na_filter='') #na_filter='' 将空值处理为'' 不然默认为NaN
for chunk in csv_data:
 rows=chunk.values.tolist()
 for x in range(0,len(rows)):                                              #将rows中的''处理为None,不然沙雕MySQL INT类型字段值''会变成0 VARCHAR字段值会变成'' 而不是NULL
    for i in range(0,len(rows[x])):
       if rows[x][i]=='':
        rows[x][i]=None
 cur.executemany(insert_sql,rows)                                          #批量插入
 conn.commit()
cur.close()
conn.close()
print('导入完成:',time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))

想不想成为SQL大神？想不想精通SQL优化，精通所有关系型数据库优化？
加我微信692162374 报名单独辅导 ， 顺带我还可以教你Python，Oracle，MySQL，大数据，以及数据仓库建模