Dask：创建严格递增的索引

Question

正如有据可查的那样，当调用reset_index 时，Dask 会在每个分区的基础上创建一个严格递增的索引，从而导致整个集合中的索引重复。在整个集合中，在 Dask 中创建严格递增的索引（不必连续）的最佳方法是什么（例如计算速度最快）？我希望map_partitions 会传递分区号，但我认为不会。谢谢。

编辑

感谢@MRocklin，我已经做到了这一点，但我需要一些帮助来了解如何将我的系列与原始数据框重新组合。

def create_increasing_index(ddf:dd.DataFrame):
    mps = int(len(ddf) / ddf.npartitions + 1000)
    values = ddf.index.values

    def do(x, max_partition_size, block_id=None):
        length = len(x)
        if length == 0:
            raise ValueError("Does not work with empty partitions. Consider using dask.repartition.")

        start = block_id[0] * max_partition_size
        return da.arange(start, start+length, chunks=1)

    series = values.map_blocks(do, max_partition_size=mps, dtype=np.int64)
    ddf2 = dd.concat([ddf, dd.from_array(series)], axis=1)
    return ddf2

我收到错误“ValueError: Unable to concatenate DataFrame with unknown Division specified axis=1”。有没有比使用 dd.concat 更好的方法？谢谢。

再次编辑

实际上，就我的目的（以及我正在测试的数据量 - 只有几 GB）而言，cumsum 已经足够快了。当这变得太慢时，我会重新访问！

Answer 1

实现这一点的一个相当慢的方法是创建一个新列，然后使用cumsum

ddf['x'] = 1
ddf['x'] = ddf.x.cumsum()
ddf = ddf.set_index('x', sorted=True)

这既不是很慢也不是免费的。

鉴于您的问题是如何表述的，我怀疑您只是希望为每个分区创建一个范围，该范围由一个非常大的值分隔，您知道该值大于最大行数。你说得对，map_partitions 没有提供分区号。您可以改为使用以下两种解决方案之一。

1. 转换为 dask.array（使用.values），使用提供块索引的map_blocks 方法，然后使用dd.from_array 转换回系列。
2. 转换为 dask.delayed 对象列表，自己创建延迟系列，然后使用dd.from_delayed 转换回 dask 系列

http://dask.pydata.org/en/latest/delayed-collections.html