向 dask 数据框添加一列，通过滚动计算它 window

Question

假设我有以下代码来生成一个虚拟的 dask 数据帧：

import pandas as pd
import dask.dataframe as dd
pandas_dataframe = pd.DataFrame({'A' : [0,500,1000], 'B': [-100, 200, 300]  , 'C' : [0,0,1.0] } )    
test_data_frame = dd.from_pandas( pandas_dataframe, npartitions= 1  )

理想情况下，我想知道向数据框添加另一列的推荐方法是什么，以懒惰的方式通过滚动 window 计算列内容。

我想到了以下方法：

import numpy as np
import dask.delayed as delay

@delay
def coupled_operation_example(dask_dataframe, 
                              list_of_input_lbls, 
                              fcn, 
                              window_size, 
                              init_value, 
                              output_lbl):

    def preallocate_channel_data(vector_length, first_components):
        vector_out = np.zeros(len(dask_dataframe))
        vector_out[0:len(first_components)] = first_components
        return vector_out

    def create_output_signal(relevant_data, fcn, window_size , initiated_vec):

       ## to be written; fcn would be  a fcn accepting the sliding window


    initiatied_vec = preallocate_channel_data(len(dask_dataframe, init_value))
    relevant_data = dask_dataframe[list_of_input_lbls]
    my_output_signal = create_output_signal(relevant_data, fcn, window_size, initiated_vec)

我在写这篇文章时，确信 dask 数据框可以让我进行一些切片：但他们不允许。因此，我的第一个选择是将计算中涉及的列提取为 numpy 数组，但它们会被急切地评估。我认为性能上的损失会很大。目前我使用 h5py 从 h5 数据创建 dask 数据帧：所以一切都是懒惰的，直到我写输出文件。

到目前为止，我只处理某一行的数据；所以我一直在使用：

 test_data_frame .apply(fcn, axis =1, meta = float)

我认为滚动没有等效的功能方法windows；我对吗？我想要 F# 中的 Seq.windowed 或 Haskell 之类的东西。非常感谢任何建议。

Answer 1

我试过通过闭包来解决。一旦我完成代码，我将 post 对一些数据进行基准测试。现在我有以下玩具示例，它似乎有效：因为 dask 数据框的应用方法似乎保留了行顺序。

import numpy as np
import pandas as pd
import dask.dataframe as dd
number_of_components = 30


df = pd.DataFrame(np.random.randint(0,number_of_components,size=(number_of_components, 2)), columns=list('AB'))
my_data_frame = dd.from_pandas(df, npartitions = 1 )


def sumPrevious( previousState ) :

     def getValue(row):
        nonlocal previousState 
        something = row['A'] - previousState 
        previousState = row['A']
        return something

     return getValue


given_func = sumPrevious(1 )
out = my_data_frame.apply(given_func, axis = 1 , meta = float)
df['computed'] = out.compute()

现在是坏消息，我试图将其抽象出来，通过这个新函数传递状态并使用任意宽度的滚动 window：

def generalised_coupled_computation(previous_state , coupled_computation, previous_state_update) :

    def inner_function(actual_state):
        nonlocal previous_state
        actual_value = coupled_computation(actual_state , previous_state  )
        previous_state = previous_state_update(actual_state, previous_state)
        return actual_value

    return inner_function

假设我们初始化函数：

init_state = df.loc[0] 
coupled_computation  = lambda act,prev : act['A'] - prev['A']
new_update = lambda act, prev : act
given_func3 = generalised_coupled_computation(init_state , coupled_computation, new_update )
out3 = my_data_frame.apply(given_func3, axis = 1 , meta = float)

尝试运行它并准备好迎接惊喜：第一个元素是错误的，可能是某些指针的问题，给定奇怪的结果。有什么见解吗？

无论如何，如果传递原始类型，它似乎可以正常工作。

更新：

解决方案是使用副本：

import copy as copy

def new_update(act, previous):
    return copy.copy(act)

现在函数的行为符合预期；当然，如果需要更耦合的逻辑，则有必要调整函数更新和耦合计算函数

向 dask 数据框添加一列，通过滚动计算它 window

Adding a column to dask dataframe, computing it through a rolling window

python

pandas

numpy

dask

rolling-computation