Dask pct_change

user_290663 0
Viewed 233

I am new to Dask. I have a pandas df with a sorted DateTimeIndex of size appox. 30M rows. I am trying to do this pandas op in dask (bc the grouping is costly in pandas)

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

In other words, group the data by tic then find sequential (periods=1) price changes, which is monotonically increasing in time (since the index is a sorted DateTimeIndex). My dask code is:

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)

However, I get the following error:

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

which i dont understand. I thought pct_change is a series function.

I am aware that a df.groupby(columns_no_index).apply(user_fn) requires shuffle, but df.groupby(columns_with_index).apply(user_fn)) is much faster. Therefore this aggregation works

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

However, the above returns a multiindex with tic as both level1 and level2 indicies, whereas we care about the DateTimeIndex being the second index (e.g. level1=tic, level2=DatetimeIndex, level3=precent_change.)

As mentioned in the comment below, if i try to write my own pct_change as in

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

I get the following error:

ValueError: Length mismatch: Expected axis has 183100 elements, new values have 183921 elements

not clear to me why

dask-dataframe dask
2 Answers

I found this to work

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

df = df.groupby('tic').apply(lambda x: x.assign(close_pct_change=(x.close.diff(1)/x.close)*100))
df  = df.compute().droplevel('tic')
user_2906630

Based on your comments, this is only an indirect solution, but it should work:

  1. split the large .gz file into a set of files based on the tic.
  2. with tic-based files, you can now have a function that loads a file and runs pandas on it.
  3. wrap the function in delayed.

Rough example:

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])
user_106935960
Related