pandas数据库mysql交互

pymysql

import pymysql
 
#打开数据库连接
db=pymysql.connect(host='1.1.1.1',port=3306,user='root',passwd='123123',db='test',charset='utf8')
cursor=db.cursor()#使用cursor()方法获取操作游标
 
sql = "select * from test0811"
 
cursor.execute(sql)
info = cursor.fetchall()
 
db.commit()
cursor.close() #关闭游标   
db.close()#关闭数据库连接

pandas中，pd.to_sql()存入数据库报错，经查询不支持mysql引擎，需下载sqlalchemy包，才能存入数据库。

sqlalchemy简介

连接数据库

在sqlalchemy中，session用于创建程序与数据库之间的会话。所有对象的载入和保存都需要通过session对象。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 链接数据库采用pymysq模块做映射，后面参数是最大连接数5
ENGINE=create_engine("mysql+pymysql://root@127.0.0.1:3306/digchouti?charset=utf8", max_overflow=5)
Session = sessionmaker(bind=engine)

session = Session()

创建映射(创建表)

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker

ENGINE=create_engine("mysql+pymysql://root@127.0.0.1:3306/digchouti?charset=utf8", max_overflow=5)

#生成一个SQLORM基类，创建表必须继承他，别问我啥意思就是这么规定的
Base = declarative_base()

class Person(Base):
    __tablename__ = 'userinfo'

    id   = Column(Integer, primary_key=True)
    name = Column(String(32))

    def __repr__(self):
        return "<Person(name='%s')>" % self.name

添加数据

#创建一个person对象
person = Person(name='张岩林')
#添加person对象，但是仍然没有提交到数据库
session.add(person)
#提交数据库
session.commit()

查询数据

#获取所有数据
session.query(Person).all()

#获取name=‘张岩林’的那行数据
session.query(Person).filter(Person.name=='张岩林').one()

#获取返回数据的第一行
session.query(Person).first()

#查找id大于1的所有数据
session.query(Person.name).filter(Person.id>1).all()

#limit索引取出第一二行数据
session.query(Person).all()[1:3]

#order by,按照id从大到小排列
session.query(Person).ordre_by(Person.id)

#equal/like/in
query = session.query(Person)
query.filter(Person.id==1).all()
query.filter(Person.id!=1).all()
query.filter(Person.name.like('%ay%')).all()
query.filter(Person.id.in_([1,2,3])).all()
query.filter(~Person.id.in_([1,2,3])).all()
query.filter(Person.name==None).all()

#and or
from sqlalchemy import and_
from sqlalchemy import or_
query.filter(and_(Person.id==1, Person.name=='张岩林')).all()
query.filter(Person.id==1, Person.name=='张岩林').all()
query.filter(Person.id==1).filter(Person.name=='张岩林').all()
query.filter(or_(Person.id==1, Person.id==2)).all()

# count计算个数
session.query(Person).count()

# 修改update
session.query(Person).filter(id > 2).update({'name' : '张岩林'})

详细：https://www.cnblogs.com/aylin/p/5770888.html

 

本地批量从指定文件夹读入数据，分行进行存储至mysql数据库中

简单利用pandas读入数据存入mysql：df.to_sql(name='表名 ', con= 连接, if_exists=append:追加，replace:删除原表，建立新表再添加,fail:无视i ndex=False：不插入索引index)

import pandas as pd
from sqlalchemy import create_engine
import os
import time

def ReadFile():
    starttime=time.time()
    file_dir = './家宽资源详情/'
    filenames = os.listdir(file_dir)
    os.path.splitext(filenames[0])
    dfs = []
    for file in filenames:
        filename, ext = os.path.splitext(file)
        if ext == '.xls' or '.xlsx':
            eachdffile = pd.read_excel(file_dir + file)
            dfs.append(eachdffile)
            print('已读入%s，历时%f秒' %(file,(time.time()-starttime)))
    data = pd.concat(dfs,axis=0,ignore_index=True)#忽视索引，否则不能切片
    return data

def Dato2Mysql(data,conn):
    #分多次导入，超过5000条会报错！
    starttime=time.time()
    for i in range(0,data.shape[0],5000):
        newdata=data.loc[i:(i+4999),]
        try:
            newdata.to_sql(name='jiakuanziyuanxiangqing', con=conn, if_exists='append', index=False)
            print('第%d行开始导入5000条数据,共历时%f秒' %(i,(time.time()-starttime)))
        except:
            print('第%d行数据导入失败' %i)

def main():
    #1.建立引擎
    engine = create_engine("mysql+mysqlconnector://{}:{}@{}/{}?charset={}".format('root', 'xinzhi', '127.0.0.1:3306', 'complaint', 'utf8'))
    conn = engine.connect()#是否相当于原生sql的consor
    #2.读本地数据
    data=ReadFile()
    # 使用原生sql语句
    # createsql = 'create table tousutest(gongdanliushuihao varchar(50) primary key not null,dierlie varchar(20))'原生语句创建表
    # conn.execute(createsql)
    # #3.存入数据库
    Dato2Mysql(data,conn)
    # conn.execute('''alter table tousutest2 modify 工单流水号 varchar(50) primary key not null''')#第一创建后修改下列表属性
    # #4.查看数据
    print(pd.read_sql(sql='select count(*) from jiakuanziyuanxiangqing',con=conn))
    conn.close()

if __name__== '__main__':
    main()