dask DataFrame 查询然后示例错误答案

【问题标题】：dask DataFrame query then sample errordask DataFrame 查询然后示例错误
【发布时间】：2017-04-24 01:51:29
【问题描述】：

我试图从read_parquet 连接DaskDataFrame，然后应用查询过滤器，然后对其进行采样以将最终数据帧大小限制为小于或等于10000。这是伪代码：

import dask.dataframe as dd

df = dd.concat([ dd.read_parquet(path, index='date').query("(col0 < 4) & (date < '20170201')")  
                 for path in files ], 
               interleave_partitions=True)
df = df.sample(float(10000) / max(10000, len(df)))
df = df.compute()

但是，它失败了：

ValueError: a must be greater than 0

Traceback
---------
  File "/opt/anaconda2/lib/python2.7/site-packages/dask/async.py", line 266, in execute_task
    result = _execute_task(task, data)  
  File "/opt/anaconda2/lib/python2.7/site-packages/dask/async.py", line 247, in _execute_task
    return func(*args2)  
  File "/opt/anaconda2/lib/python2.7/site-packages/dask/dataframe/methods.py", line 143, in sample
    return df.sample(random_state=rs, frac=frac, replace=replace)  
  File "/opt/anaconda2/lib/python2.7/site-packages/pandas/core/generic.py", line 2644, in sample
    locs = rs.choice(axis_length, size=n, replace=replace, p=weights)  
  File "mtrand.pyx", line 1391, in mtrand.RandomState.choice (numpy/random/mtrand/mtrand.c:16430)

如果我不执行.query(...) 部分，那么它可以正常工作。如果我在示例之后应用查询，也可以，但是我无法控制最终的DataFrame 大小。我在这里尝试做的事情有什么问题吗？

我正在运行 OS X 10.10.5、fastparquet 0.0.5、dask 0.14.1、python 2.7.12。

【问题讨论】：

您的.query 是否返回空数据框？使用pd.DataFrame().sample() 可以从pandas 中得到类似的错误，所以我会在查询后检查len(df)。
数据没问题，len(df) 是百万。
TomAugspurger 的评论是正确的，虽然长度检查不应该在 daskdataframe 查询之后，而是在 dask.dataframe.sample(...) 函数内部（见下面的答案），因为它更常见有些分区是空的，但整个 dask.dataframe 不是空的。

标签： python parquet dask

【解决方案1】：

由于某些 pandas DataFrame 是空的，因此引发了“ValueError: a must be greater than 0”错误。这个 ValueError 是由pandas.DataFrame.sample 方法抛出的。因为我们在dask查询之后做sample，并不是所有查询的子任务都会产生一个非空的pandas.DataFrame，这个ValueError几乎肯定会发生。

正确的修复应该在dask.dataframe代码中：如果df为空，则返回它本身，否则调用df.sample：

> /opt/anaconda2/lib/python2.7/site-packages/dask/dataframe/methods.py(166)sample()
164 def sample(df, state, frac, replace):
165     rs = np.random.RandomState(state)
--> 166     return df.sample(random_state=rs, frac=frac, replace=replace)

i.e. return df.sample(random_state=rs, frac=frac, replace=replace) \
            if len(df) > 0 else df

【讨论】：