dask - 对 dask 数据帧样本的计算比对所有数据的计算要长得多

Question

我有一个 dask 数据框，由 parquet 支持。它有 1.31 亿行，当我对整个帧执行一些基本操作时，它们需要几分钟。

df = dd.read_parquet('data_*.pqt')
unique_locations = df.location.unique()
https = unique_locations.str.startswith('https:')
http = unique_locations.str.startswith('http:')
total_locations = len(unique_locations)
n_https = https.sum().compute()
n_http = http.sum().compute()

时间:

CPU times: user 2min 49s, sys: 23.9 s, total: 3min 13s
Wall time: 1min 53s

我天真地认为，如果我对数据进行抽样，我这次可以把它拿下来，然后做了:

df = dd.read_parquet('data_*.pqt')
df = df.sample(frac=0.05)
unique_locations = df.location.unique()
https = unique_locations.str.startswith('https:')
http = unique_locations.str.startswith('http:')
total_locations = len(unique_locations)
n_https = https.sum().compute()
n_http = http.sum().compute()

时间:

Unknown, I stopped it after 45minutes.

我猜测我的示例无法在我所有的后续计算中有效访问，但我不知道如何修复它。

我对从 dask 数据帧中采样数据然后处理该样本的最佳方法很感兴趣。

Answer 1

我没有明确/简单的答案，但我确实有很多东西可以一起解决我的问题。

1) 我的代码效率低下，挑选出我需要处理的特定列使一切正常。我的新代码:

import dask.dataframe as dd
from dask.distributed import Client, progress
client = Client()  # Took me a little while to get the settings correct

def get_df(*columns):
    files = '../cache_new/sample_*.pqt'
    df = dd.read_parquet(files, columns=columns, engine='pyarrow')
    return df

# All data - Takes 31s
df_all = get_df('location')
unique_locations = df_all.location.unique()
https = unique_locations.str.startswith('https:')
http = unique_locations.str.startswith('http:')
_total_locations = unique_locations.size.persist()
_n_https = https.sum().persist()
_n_http = http.sum().persist()
progress(_total_locations, _n_https, _n_http)

# 1% sample data - Takes 21s
df_sample = get_df('location').sample(frac=0.01)
unique_locations = df_sample.location.unique()
https = unique_locations.str.startswith('https:')
http = unique_locations.str.startswith('http:')
_total_locations = unique_locations.size.persist()
_n_https = https.sum().persist()
_n_http = http.sum().persist()
progress(_total_locations, _n_https, _n_http)

事实证明这并不是一个很大的加速。整个计算所花费的时间主要是读入数据。如果计算非常昂贵，我想我会看到更多的加速。

2) 我切换到在本地使用分布式调度程序，这样我就可以看到发生了什么。但这并非没有问题:

1. 我遇到了 fastparquet 的某种错误，导致我的进程死了，我需要使用 pyarrow(这在不使用分布式客户端时不是问题)
2. 我不得不手动设置线程数和内存限制

3) 我发现了在笔记本中多次读取相同数据的错误 - https://github.com/dask/dask/issues/3268

4) 我也遇到了 pandas 中的内存泄漏错误 https://github.com/pandas-dev/pandas/issues/19941#issuecomment-371960712

有了 (3) 和 (4) 以及在我的原始代码中我阅读所有列的效率低下这一事实，我看到了很多原因为什么我的示例没有工作，尽管我从未找到明确的答案。