在dask数据框中添加一列,并通过滚动窗口对其进行计算 - python

假设我有以下代码,生成一个虚拟的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  )

理想情况下,我想知道什么是推荐的方法,即以一种懒惰的方式通过滚动窗口计算该列的内容,从而将另一列添加到数据框中。

我想出了以下方法:

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)

我认为滚动窗口没有等效的功能方法。我对吗?我想要F#或Haskell中的Seq.windowed之类的东西。任何建议高度赞赏。

参考方案

我试图通过关闭来解决它。完成代码后,我将在一些数据上发布基准测试。现在,我有一个下面的玩具示例,该示例似乎起作用:由于dask数据框的apply方法似乎保留了行顺序。

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()

现在,坏消息是,我试图通过以下新函数将其抽象出来,传递状态并使用任意宽度的滚动窗口:

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)

现在,这些函数的行为符合预期;当然,如果需要更耦合的逻辑,则有必要调整函数更新和耦合的计算功能

Python-Excel导出 - python

我有以下代码:import pandas as pd import requests from bs4 import BeautifulSoup res = requests.get("https://www.bankier.pl/gielda/notowania/akcje") soup = BeautifulSoup(res.cont…

Python:如何根据另一列元素明智地查找一列中的空单元格计数? - python

df = pd.DataFrame({'user': ['Bob', 'Jane', 'Alice','Jane', 'Alice','Bob', 'Alice'], 'income…

R'relaimpo'软件包的Python端口 - python

我需要计算Lindeman-Merenda-Gold(LMG)分数,以进行回归分析。我发现R语言的relaimpo包下有该文件。不幸的是,我对R没有任何经验。我检查了互联网,但找不到。这个程序包有python端口吗?如果不存在,是否可以通过python使用该包? python参考方案 最近,我遇到了pingouin库。

如何将熊猫数据框合并到现有的reportlab表中? - python

example_df = [[1,2,3,4,5],[1,2,3,4,5],[1,2,3,4,5]] 我想将example_df pandas数据框集成到现有的Reportlab表中-行数正在变化(如示例中所示,可以是3,也可以是20):rlab_table(['Mean','Max','Min',…

熊猫货币转换 - python

我在下面的代码段遇到麻烦。我的系统一旦开始执行下面的第三行,就会用完内存。我试图根据'M / D'列将金额转换为USD,该列表示是否需要乘以'LDIVND'金额或将'Rate'除以。 “ USDEQ”是我尝试创建并存储转换后金额的新列。任何帮助表示赞赏。t['M/D']=t['M/D'].astype(str) t.l…