Python连接数据库、查询、建表、上传和下载数据

前言：

此文利用python脚本连接mysql，使用本地csv/xlsx 文件，进行建表、上传、下载、更新数据等操作

一、导入库

# Singghet
import pandas as pd
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
import pymysql

二、连接数据库并读取下载数据

db = pymysql.connect(host="localhost", user=root, password=123456, port=3308, database=sql_test)

参数：

注1：这里遇到的坑就是没有写端口号报错，建议默认端口也设置port参数

下载数据

sql = select * from dau;
df = pd.read_sql_query(sql, con=db)

这样就成功读取成DataFrame到本地了

三 创建表并查询、上传至数据库、导出至本地

现在用读取到的数据，筛选出每个用户的最大连续登陆天数，然后导出至本地，并且在数据库中也创建一个用户最大连续表

# 初始化Base表
Base = declarative_base()


class Maxcday(Base):
    __tablename__ = max_continue_day
    # 定义字段
    uid = Column(Integer, primary_key=True)   #将用户id设置为主键
    max_c_day = Column(Integer)               # 为最大连续登陆天数


# 将数据写入mysql的数据库，但需要先通过sqlalchemy.create_engine建立连接,且字符编码设置为utf8，否则有些latin字符不能处理
engine = create_engine(mysql+pymysql://root:123456@localhost:3308 /sql_test?charset=utf8)
# 创建表
Base.metadata.create_all(engine)

如果要取最大登陆天数，可以直接使用sql的窗口函数查询

注2 ： 以下是查询的SQL脚本，无法直接在python中运行

SELECT c.uid,MAX(c.max_c_day) as max_c_day
FROM
	(SELECT b.uid,COUNT(0) AS max_c_day
	FROM(
		SELECT a.*,DATE_SUB(a.dt,INTERVAL a.d_rank DAY) as diff
		FROM (
			SELECT DISTINCT uid, dt,DENSE_RANK()OVER (PARTITION BY uid ORDER  BY dt ASC) AS d_rank
			FROM dau) a
			)b
	GROUP BY b.uid,b.diff)c
GROUP BY c.uid ；

在navicat的查询结果如下： python中实现上图的sql查询

sql_script = SELECT c.uid,MAX(c.max_c_day) as max_c_day FROM	(SELECT b.uid,COUNT(0) AS max_c_day FROM(SELECT a.*,DATE_SUB(a.dt,INTERVAL a.d_rank DAY) as diff FROM (SELECT DISTINCT uid, dt,DENSE_RANK()OVER (PARTITION BY uid ORDER  BY dt ASC) AS d_rank FROM dau) a)b GROUP BY b.uid,b.diff)c GROUP BY c.uid;
df_max_c_day = pd.read_sql_query(sql=sql_script, con=engine)       # 设置sql脚本和连接
df_max_c_day.to_csv(max_c_day_table.csv, index=False)             # 导出Dataframe为csv文件
df_max_c_day.to_sql(max_continue_day, engine, schema=sql_test, if_exists=append, index=None)   # 参数依次为 导出的表、连接，数据库，输出形式为追加输出

这样将查询结果导出为csv文件，并且将查询结果导入到本地mysql中，先前创建好的max_continue_day表中了 （PS: df.to_sql()也方法，也可以直接将本地csv/xlsx文件或者从API接口中得到的Dataframe型数据导入至数据库中）

四、结束总结

使用pymsql和sqlalchemy库结合着pandas导出dataframe至数据库的方法，如果后期需要追加导入数据，可以将原先数据读出和现有数据进行上下合并，再重新导入，方式选择‘replace’替换即可；若数据量比较大，可以先导入再直接使用sql语句进行去重操作。