AttributeError：“系列”对象在 Dask 中没有属性“列”答案

【问题标题】：AttributeError: 'Series' object has no attribute 'columns' in DaskAttributeError：“系列”对象在 Dask 中没有属性“列”
【发布时间】：2019-05-15 20:51:33
【问题描述】：

我有一个函数应该应用于某些数据框以进行一些计算。由于数据框非常大，旨在加快计算速度，我决定选择 Dask 进行并行 pandas 处理

我有以下代码： https://pastebin.com/Zh672Wei

主要问题在于以下几行：

crosses_data.apply((lambda row: calculate_vwap(row[0], row[1], row[2], vwap_data, row.name)), axis=1)

上面的代码有效。相同的代码，但与 dask 并行失败，并出现错误“系列”对象没有属性“列”：

dd.from_pandas(crosses_data,npartitions=4).map_partitions(
      lambda df : df.apply((lambda row: calculate_vwap(row[0], row[1], row[2], vwap_data, row.name)), axis=1)).\
   compute(scheduler=get)

我使用 dask 的官方文档，这个错误现在真的很有意义。

【问题讨论】：

标签： python pandas parallel-processing dask

【解决方案1】：

可能是某种魔法，但解决方案如下：。 calculate_vwap 函数应该有一个 touple 作为输出：

def calculate_vwap(ric_id, interval_start, interval_finish, vwap_data, row_n):
    some_tmp_vwap_interval_data = \
        vwap_data.query(
            'TKER == @ric_id and interval > @interval_start and interval < '
            '@interval_finish '
        )[['IVWP', 'INTV']]
    if sum(some_tmp_vwap_interval_data['INTV']):
        return \
            sum(
                vwap * volume for vwap, volume in
                zip(some_tmp_vwap_interval_data['IVWP'],
                    some_tmp_vwap_interval_data['INTV'])
            ) \
            / sum(some_tmp_vwap_interval_data['INTV']), \
            some_tmp_vwap_interval_data.IVWP.iloc[0], \
            some_tmp_vwap_interval_data.IVWP.iloc[-1], \
            some_tmp_vwap_interval_data.INTV.sum()

    return None

之后，此输出应转换为数据帧：

pd.DataFrame(
                dd.from_pandas(crosses_data[[
                    'RIC', 'Interval_Start_Human',
                    'Interval_End_Human']],
                               npartitions=int(partitions_number)).
                map_partitions(
                    lambda df: df.apply((
                        lambda row: calculate_vwap(row[0],
                                                   row[1],
                                                   row[2],
                                                   vwap_data,
                                                   row.name)),
                                        axis=1)).
                compute(scheduler=get).values.tolist())

【讨论】：