记一次用dataframe进行数据清理

总结一下dataframe读取数据库，以及整理数据的过程。分为三个部分：数据读取，数据整理以及数据写入。

1、数据读取

从csv读取读取数据，使用pandas读的read_csv函数，传入两个参数，分别是path文件路径，usecols读取的列表，返回的是dataframe格式。

import pandas as pd
 
def csv_read(path, usecols):
    return pd.read_csv(path, usecols=usecols, encoding="gbk")

2、数据整理

我需要做清除空值，替换固定值，匹配关键词等操作。

1）清除空值很简单，按列替换空值，效率挺快。

df.fillna(0, inplace=True)

2）替换固定值很简单， 按列替换空值，效率挺快

df["s_serialno"] = df["s_serialno"].str.replace("'", "")

3）匹配关键词，由于多个关键词对多个内容，使用第一点的按列匹配就不行，我使用遍历方法

itertuples，其他的遍历方法可参见：Python - pandas DataFrame数据的合并与拼接（merge、join、concat）_pd.merge合并后顺序-CSDN博客

 一开始我是匹配到不到关键词，就直接删除掉，测试小量数据还行，但面对几十万数据，频繁地让Dataframe删除数据，效率很低。于是作了如下改写，先将需要删作的idx保存下来，再一并删除。

dropindex = []
keyword= 'XXXXX'
 
for obj in df.itertuples():
     idx = getattr(obj, "Index")
     if getattr(obj, "cnt") not in keyword.to_string():
            dropindex.append(idx)
 
df.drop(dropindex, inplace=True)

3、数据写入

使用dataframe.to_sql方法，开始的写法，为保证数据唯一性，避免主键重复出错，使用try except方法 ，一条一条录入，若主键重复直接pass即可。小量数据测试还行，面对几十万数据，单条录入，数据库与程序的I/O效率不高，数据录入缓慢。

from sqlalchemy import create_engine
 
def mysql_engine():
    return create_engine("mysql+pymysql://root:10086@192.168.1.1:3306/order")
 
def write_sql(rows):
     # 使用逐条插入，而不用批量插入，用try-except判断避免重复插入的异常
     for i in range(len(df)):
         try:
             df.iloc[i : i + 1].to_sql(
                 name=table_name, con=mysql_engine(), if_exists="append", index=False
             )
         except Exception as e:
             # print(e)
             pass

于是我改用另一种方法：

使用原生sql语句，批量导入数据，使用ON DUPLICATE key UPDATE 避免主键重复出错。

import pymysql
 
def getConn():
    return pymysql.connect(
        host="192.168.1.1", user="root", password="123456", database="order"
    )
 
def write_sql(rows):
    sql = (
        "insert into order.TABLE("
        "s_A,"
        "s_B"
        ") values(%s,%s)"
        "ON DUPLICATE key UPDATE s_B=values(s_B)"
    )
    conn = getConn()
    cur = conn.cursor()
    cur.executemany(sql, rows)
    conn.commit()
    cur.close()
    conn.close()