多个处理相同查询的熊猫df然后将df合并成单个df

Question

我想总结一下许多表的记录计数，我希望并行运行以节省时间，。

list_tabl_cn =['TBL_A', 'TBL_B', 'TBL_C', 'TBL_D']

def tblRowCn(p_tbl):
    #connDb = pyodbc.connect(f'DSN={nama_db_target}', autocommit =True)
    connDb = sqlanydb.connect(uid='dba', 
                               pwd='sql',
                               host='ip:port', 
                               dbn='blah')    
    is_tableExists = ego.my_desc(p_tbl,163).shape[0] 
    if is_tableExists:
        proc_name = 'df_'+p_tbl
        if p_tbl == 'STG_CFG_SYS':
            Q_ = """\
              SELECT OPENDATE as TGL_POS, COUNT(1) CN FROM {0}
              GROUP BY TGL_POS
            """.format(p_tbl)
        else:
            Q_ = """\
              SELECT TANGGAL_POSISI as TGL_POS, COUNT(1) CN FROM {0}
              GROUP BY TGL_POS
            """.format(p_tbl)
        df_tbl = pd.read_sql_query(Q_, connDb, parse_dates=['TGL_POS'])
        df_tbl['THN'],df_tbl['BLN']= df_tbl['TGL_POS'].dt.year, df_tbl['TGL_POS'].dt.month    
    else:
        df_tbl=[]

    return df_tbl

def task(table_nm):
    print(f"Task Executed with process {mp.current_process().pid}")
    tblRowCn(table_nm.upper())

def main():
    executor = mp.Pool(mp.cpu_count()-8)
    executor.map(task, [n_table_nm for n_table_nm in list_tabl_cn])
    executor.close()

if __name__ == "__main__":
    main()  

可能是这样的

def main():
    executor = mp.Pool(mp.cpu_count()-8)
    executor.map(task, [n_table_nm for n_table_nm in list_tabl_cn])
    append.[task1, task2, task...]   
    executor.close()

我的全部数据就是append.[task1, task2, task...]

我相信我在代码中遗漏了一些东西，但它太模糊了。

Answer 1

如果您的所有Dataframes都有相同的列，并且希望将所有数据文件的所有行都添加到一个dataframe中，那么您可以使用熊猫concat函数。将所有单独的数据添加到一个列表中，然后将所有这些数据连接起来，以生成您的主数据。

list_of_df =[] 
for df in executor.map(task, list_tabl_cn):
    if df:
        list_of_df.append(df)
main_df = pd.concat(list_of_df)

您可以在tblRowCn方法中删除它的其他条件，它是多余且不需要的。

在您的代码中，您从list_tabl_cn生成了一个列表来将它传递给map函数，您不需要这样做，您可以像上面的代码那样直接给list_tabl_cn映射函数。