如何创建作为Dask数据帧子集的新Dask数据帧？

Question

如何应用基于另一个dask数据帧的值创建新dask数据帧的函数？我有一个函数，它根据原始数据帧函数的值返回一个新的数据帧。我已将此数据帧改编为具有多个分区的dask数据帧：

df = dd.from_pandas(old_df, npartitions=4)

给定一个旧函数：

def create_ndf(row):
    a = [row.a, row.b, row.c*row.d]
    n_df = pd.DataFrame(columns=['A','B','Mult'], data=a)
    return n_df

我将上述功能调整为：

n_df = dd.from_pandas(pd.DataFrame(), npartitions=4)

def create_ndf(row, new_dataframe):
    a = pd.DataFrame(columns=['A','B','Mult'],data=[row.a,row.b,row.c*row.d])
    new_dataframe = dd.concat([new_dataframe, a], axis=0, interleave_partitions=True)
    return new_dataframe


n_df = df.map_partitions(lambda x: x.apply(lambda row: create_ndf(row, n_df),axis=1)).compute()

但是，上面的代码不起作用。n_df的最终输出是一个StopIteration。我在更新的create_ndf函数中放入了一个print语句来检查数据帧分配的输出，new_dataframe的值不会增加它拥有的分区的数量，但它会获取新输入的数据帧的值