一个读取excel数据处理完成后读入数据库的例子

最近收集了一批数据，各地根据问题数据做出反馈，但是各地在反馈的时候字段都进行了创新，好在下发的数据内容并没有改变，开始写的单进程的，由于时间较长，耗时380+秒，又改成多进程的，时间缩短为80-秒。现在把程序发出来，请各位大神进行指正。

 import multiprocessing
import os
import time
import pandas as pd
from sqlalchemy import create_engine
import asyncio
import  warnings

# warnings.simplefilter("ignore")
ywstrList=['经办机构', '原子业务编号', '原子业务名称', '风险名称','风险描述',
           '校验规则结果', '创建时间', '风险提示信息', '业务日志号']
ywstrListMemo=['经办机构', '原子业务编号', '原子业务名称', '风险名称','风险描述',
           '校验规则结果', '创建时间', '风险提示信息', '业务日志号','memo','time']
szlist = ['省本', '成都', '自贡', '攀枝', '泸州', '德阳', '绵阳', '广元',
              '遂宁', '内江', '乐山', '南充', '眉山', '宜宾', '广安', '达州',
              '雅安', '巴中', '资阳', '阿坝', '甘孜', '凉山']
# connect = create_engine('mysql+pymysql://root:@127.0.0.1:3306/ywgk?charset=utf8')
connect = create_engine('mysql+mysqlconnector://root:@127.0.0.1:3306/ywgk?charset=utf8')
# engine = create_engine('mysql+mysqlconnector://scott:tiger@localhost/foo')

def findSZ(filename):
    for sz in szlist:
        if filename.find(sz) != -1:
            return sz
    return None


def ReadExcel(filename):
    xlsdf = ''
    xlsdf = pd.read_excel(filename)
    """"
    remove columns='sz' or '市'
    """
    if "市" in list(xlsdf.keys().to_list()):
        xlsdf.drop(columns='市', axis=1, inplace=True)
    if "sz" in list(xlsdf.keys().to_list()):
        xlsdf.drop(columns='sz', axis=1, inplace=True)
    xlsdf = xlsdf.fillna("").astype('string')
    return xlsdf


def filterDataOfSz(filename, xlsdf):
    sz = findSZ(filename)
    print(sz)
    """
    筛选出包含对应市州的数据。
    """
    if sz != None:
        xlsdf = xlsdf[xlsdf['经办机构'].str[:2] == sz]  # 筛出本市州数据
        return xlsdf


def ConCatRestCols(xlsdf):
    """
    去掉业务部分字段，保留市州反馈意见。
    """
    print(xlsdf)
    # if xlsdf==None:
    #     print(filename+"为空，需要处理")
    #     return
    xlfdf_keys_set = set(xlsdf.keys().to_list())
    xlsdf_restkeys_set = xlfdf_keys_set - set(ywstrList)
    xlsdf_restkeys_list = list(xlsdf_restkeys_set)
    xls_rest_df = xlsdf.loc[:, xlsdf_restkeys_list]  # 可以正确操作
    xlsdf['memo'] = '#'
    for col in xlsdf_restkeys_list:
        xlsdf['memo'] += xlsdf[col]

    #
    return xlsdf


def SetTimeStamp(filename, xlsdf):
    xlsdf['time'] = os.stat(str(filename)).st_mtime
    return xlsdf

async def ProcessExcelAndtosql(filename, table):
    df = ReadExcel(filename)
    df = filterDataOfSz(filename=filename, xlsdf=df)
    df = ConCatRestCols(df)
    df = SetTimeStamp(filename=filename, xlsdf=df)
    df = df.loc[:, ywstrListMemo]
    print(filename)
    print(df)
    df.to_sql(name=table, con=connect, if_exists='append', index=False, chunksize=1000, method='multi')
def profile(func):
    def wrapers(*args,**kwargs):
        print("测试开始")
        begin=time.time()
        func(*args,**kwargs)
        end=time.time()
        print(f"耗时{end-begin}秒")
    return wrapers

# async def getmsg(msg):
#     print(f'#{msg}')
#     await asyncio.sleep(1)

def getFiles(src:str):
    import pathlib
    files=[]
    for file in  pathlib.Path(src).rglob("*.xls?"):
        files.append(str(file))
    return files
def process_asyncio(files,table):
    loop=asyncio.new_event_loop()
    tasks=[loop.create_task(ProcessExcelAndtosql(filename,table)) for filename in files]
    loop.run_until_complete(asyncio.wait(tasks))
@profile
def run(iterable,table):
    process_count = multiprocessing.cpu_count()

    # print(process_count)
    pool = multiprocessing.Pool(process_count-2)
    iterable=get_chunks(iterable, process_count)
    for lst in iterable:
        pool.apply_async(process_asyncio, args=(lst,table))
    pool.close()
    pool.join()
def main():
    files = getFiles(r"e:市州返回")
    run(files, 'ywgk3')

def get_chunks(iterable,num):
    # global iterable
    import numpy as np
    return np.array_split(iterable, num)

# import  profile
if __name__=="__main__":    
    main()                   

本人只是编程的业余爱好者，只是把技术用于辅助工作，并没有深入研究技术理论，都是野路子，还请批评指正。