Dask - 使用 itertuples 按顺序读取分区答案

【问题标题】：Dask - Reading partitions in order using itertuplesDask - 使用 itertuples 按顺序读取分区
【发布时间】：2017-11-15 19:02:00
【问题描述】：

我正在使用 Dask 读取使用 read_sql_table 的大约 1400 万行的表。当我使用itertuples 读取数据帧时，索引（在表中排序）不会按顺序读取一个或两个分区。怎么可能强制执行呢？ row_id 由row_number（在视图上）生成，并在生成数据帧时用作索引。我知道 Pandas 有一个 sorted=True arg，有什么类似的吗？

这是在读取数据时发生的情况（读取的行数应与当前索引匹配）： INFO - Read 11870000 Rows (index: 11870000) INFO - Read 11880000 Rows (index: 11880000) INFO - Read 11890000 Rows (index: 11890000) INFO - Read 11900000 Rows (index: 11900000) --INFO - Read 11910000 Rows (index: 12159912)-- INFO - Read 11920000 Rows (index: 12169912) INFO - Read 11930000 Rows (index: 12179912) INFO - Read 11940000 Rows (index: 12189912)

在第 11,900,000 行之前一切正常，此时它切换到错误的分区。

【问题讨论】：

你能发一个stackoverflow.com/help/mcve吗？它是特定于 SQL 的吗？你说read_sql_table，是dask还是pandas？（只是好奇：你为什么使用itertuples？这对于 1400 万条记录来说会很慢）
这是为了输入第三方建模工具，因此有多种选择（包括读取数据、以某种方式更改数据以及将其写回数据库）。在这种情况下，我想看看性能如何，而只是将其保存在 Dask DF 中。我不确定 MVCE 是否会突出这个问题，因为分区是按顺序正确创建的，但是当使用 itertuples（在 Dask 中）进行迭代时，看起来分区是以不同的顺序加载的；但只是偶尔，这让我觉得请求在某种程度上是并行的。

标签： python pandas dask

【解决方案1】：

这可能是问题的答案（这可能很少见），但读取流的软件需要单调递增的索引。我只能假设它是以不同的速度解决的对数据库的多次调用，因此可能另一种选择是在compute 对read_sql_table 的调用上使用单线程调度程序。

首先，我得到每个分区的第一个索引；

def _order_partitions(self, ddf):
    ordering = {}
    for partition in range(ddf.npartitions):
        ordering.update({partition: int(df.get_partition(partition).head(1).index[0])})

    return sorted(ordering, key=ordering.get)

将结果存储在self._ordered_partitions，然后我在 Dask 中重新创建 itertuples 函数调用（这很简单）；

def _generator(self):
    for i in range(self._ddf.npartitions):
        ordered_partition = self._ordered_partitions[i]
        df = self._ddf.get_partition(ordered_partition).compute()
        for row in df.itertuples():
            yield row

唯一的变化是添加了ordered_partition。我还没有完全测试它，所以一旦我对它感到满意就会标记为答案。

【讨论】：