pandas 高效 groupby 将两个数据帧与 tqdm 一起应用

Question

方法一（清晰但很慢）

product_ids = df1.product_id.unique()
store_ids= df1.store_id.unique()

with tqdm(total=product_ids.shape[0]*store_ids.shape[0]) as t:
    for product_id in product_ids:
        p1 = df1.loc[(df1.product_id==product_id)]
        p2 = df2.loc[(df2.product_id==product_id)]
        for store_id in store_ids:
            df11 = p1.loc[(p1.store_id==store_id)]
            df22 = p2.loc[(p2.store_id==store_id)]
            train_predict(df11, df22)
            t.update()

方法2（快但我不喜欢）

df1 = df1.reset_index()
df2 = df2.reset_index().set_index(['store_id', 'product_id'])

def _reduce(df_orderitems):
    MIN_ORDERITEMS_COUNT = 30

    store_id = df_orderitems.store_id.iloc[0]
    product_id = df_orderitems.product_id.iloc[0]

    try:
        ## !!!! here refer to global df2, I don't like  !!!!!
        df_stockquantitylog = df2.loc[(store_id, product_id)]
        ## !!!! here refer to global df2, I don't like  !!!!!   
    except KeyError:
        logger.info('## df_orderitems shape:%s , cannot find (%s, %s)' % (df_orderitems.shape, store_id, product_id) )
        return

    train_predict(df_orderitems, df_stockquantitylog)

tqdm.pandas()
df1.groupby(['store_id', 'product_id']).progress_apply(_reduce)

我需要 tqdm 来显示进度条，但是 Method1 很慢（我认为是因为打印效率低下）。方法2有tqdm的pandas补丁，我认为另一个关键点是groupby.apply。但我不知道如何让方法 1 和方法 2 一样快。

注意：

df1.shape[0] != df2.shape[0] ，无法合并。 它们是从数据库中转储的。例如，df1 中可能有 10 行 store_id A 和 product_id B 相同，df2 中可能有 100 行 store_id A 和 product_id B 相同。在正确处理之前不能合并它们：

需要：

1. 首先按 store_id 和 product_id 选择（在每个 df1 和 df2 中）
2. 没有选择就无法加入。我必须对df1[(df1.store_id==A)&(df1.product_id==B)]) 和df2[(df2.store_id==A)&(df2.product_id==B)]) 应用不同的聚合，以便为它们提供相同的DatatimeIndex 进行合并，因为某些元数据列需要按日期聚合。您不能在没有选择的情况下执行此操作，因为 store_id 和 product_id 的不同组合具有重复的日期。
3. 那么这两个结果是mergable(joinable)
4. 训练模型

Answer 1

因此，如果您对方法 2 的唯一关注是您不想查看全局 df2，为什么不将其作为第二个参数传递给您的函数呢？ 例如。

def _reduce(df_orderitems, df_): ...

但我不建议完全按照您在此处概述的任何一种方法进行。

像在方法 1 中那样迭代数据帧永远不会像使用 apply 那样快，因为 apply 是使用 Cython 进行内部优化的。事实上，它会非常慢（正如你所发现的那样）。 Here 很好地解释了为什么以及您加快操作的选择。

你的问题有点含糊，你为什么不喜欢方法 2，但如果我站在你的立场上，我会做两件事。

1. 我会在 pandas 中使用类似 SQL 的操作将两个数据帧连接在一起。

您在 df1 和 df2 中都有重叠列（“store_id”和“product_id”），所以我会在 pandas 中使用 SQL 样式的 join 来组合这两个数据框。这样您就不必处理当前正在执行的索引位。

让我们首先创建一些我认为代表您的情况的虚拟数据：

  df1 = pd.DataFrame({"store_id": ['A','A','A','B','B'],
                      "product_id": [0, 1, 2, 1, 0],
                      "record_number": [0, 1, 2, 3, 4], 
                      "data": [21, 22, 28, 26, 25]})

  df2 = pd.DataFrame({"store_id":['A','A','A','B','B', 'B'],
                      "product_id": [0, 1, 2, 0, 1, 2], 
                      "more_data":[35, 39, 36, 33, 37, 32]})

然后您可以使用join 函数在重叠列上合并两个数据框，SQL 样式（join 使用数据框的索引）。这会将 df2 中的数据映射到 df1 中的数据，以创建一个新的合并数据框。 (Described in detail in the pandas docs)

 merged = df1.join(df2.set_index(['store_id','product_id']), 
                   how='left', 
                   on=['store_id','product_id'], 
                   rsuffix='_df2_data')

这给了你

    store_id  product_id  record_number  product_data  more_product_data
  0        A           0              0            21                 35
  1        A           1              1            22                 39
  2        A           2              2            28                 36
  3        B           1              3            26                 37
  4        B           0              4            25                 33

根据您的数据完整性，您可能需要检查 NaN

您还可以根据需要过滤、转换等数据（如果您的处理步骤还有其他步骤）

1. 然后我会使用apply (progress_apply) 进行预测步骤

您可以创建一个新函数，也可以使用 lambda 函数，具体取决于您的情况

def train_predict(a, b): 
    return a + b 

def predict_outcome(df_row):     
    return train_predict(df_row[['product_data']].values[0], 
                         df_row[['more_product_data']].values[0])

tqdm.pandas(desc='predict the outcome')
merged['prediction'] = merged.progress_apply(lambda x:  train_predict(x['product_data'],x['more_product_data']),
                                             axis='columns')

# or 

tqdm.pandas(desc='predict the outcome')
merged['prediction'] = merged.progress_apply(predict_outcome, axis='columns')

（axis='columns' 参数告诉apply 遍历df 的行）