Dask pct_change

Question

我是 Dask 的新手。我有一个 Pandas df，它的排序 DateTimeIndex 大小约为。 30M 行。我正在尝试在 dask 中执行此 Pandas 操作(bc 分组在 Pandas 中很昂贵)

df0['close_pct_change'] = df0.groupby('tic').close.pct_change(periods=1).dropna()

换句话说，按 tic 对数据进行分组然后找到连续(周期 = 1)价格变化，它在时间上单调增加(因为索引是一个排序的 DateTimeIndex)。我的 dask 代码是:

import dask
from dask.distributed import Client,progress
import dask.dataframe as dd

client = Client(n_workers=4, threads_per_worker=2, memory_limit='8GB')

df = dd.from_pandas(df0, npartitions=10)
df = df.persist()

df.groupby('tic', dropna=True).close.pct_change(1)

但是，我收到以下错误:

AttributeError: 'SeriesGroupBy' object has no attribute 'pct_change'

我不明白。我以为 pct_change是一个系列函数。
我知道 df.groupby(columns_no_index).apply(user_fn)需要洗牌，但 df.groupby(columns_with_index).apply(user_fn))快得多。因此这种聚合有效

dask_job = df0 \
    .set_index('tic')\
    .pipe(dd.from_pandas, npartitions=20)\
    .groupby(['tic']) \
    .apply(lambda x: x.close.pct_change(1)) 

但是，上面返回一个带有 tic 的多索引。作为 level1 和 level2 索引，而我们关心 DateTimeIndex 作为第二个索引(例如 level1=tic、level2=DatetimeIndex、level3=precent_change。)
正如在下面的评论中提到的，如果我尝试写我自己的 pct_change如

df.groupby('tic',dropna=True).close.transform(lambda x: x.diff(1)/x, meta=('close_pct_change', 'f8')) 

我收到以下错误: ValueError: Length mismatch: Expected axis has 183100 elements, new values have 183921 elements我不清楚为什么

Answer 1

根据您的评论，这只是一个间接解决方案，但它应该有效:

拆分大.gz文件转换成一组基于tic的文件.

与 tic基于文件，您现在可以拥有一个加载文件并在其上运行 Pandas 的函数。

将函数包装在延迟中。

粗略的例子:

from dask import delayed
@delayed
def process_file(path_to_file):
    df = pd.read_csv(path_to_file) # parquet is probably better, but that's not important
    df['close_pct_change'] = df.groupby('tic').close.pct_change(periods=1).dropna()
    return df

results = dask.compute([process_file(f) for f in file_list])