重新划分Dask数据框架，使用自定义索引

Question

我有一个巨大的Dask数据帧与此类似

|Ind| C1 | C2 |....| Cn |
|-----------------------|
| 1 |val1| AE |....|time|
|-----------------------|
| 2 |val2| FB |....|time|
|-----------------------|
|...|....| .. |....| ...|
|-----------------------|
|n-1|valx| ZK |....|time|
| n |valn| QK |....|time|

我想根据C2列的唯一值重新划分它，并将一个函数映射到每个分区。在

首先，我将C2设为索引：

^{pr2}$

现在我想重新划分新索引的数据帧，并将一个函数映射到每个分区。我目前的做法是这样的：

unique_c2 = df.index.unique().compute()

df = df.repartition(division=list(unique_c2))

# list(unique_c2) looks like this: ['AE', 'FB', ..., 'ZK', 'QK']

df.map_partitions(lambda x: my_func(x), meta=df)

我想要的分区应该如下所示：

|Ind | C1 | C2 |....| Cn |
|------------------------|
| AE |val1| AE |....|time|
|------------------------|
| AE |val2| AE |....|time|
|------------------------|
|....|....| .. |....| ...|
|------------------------|
| AE |valn| AE |....|time|

...

|Ind | C1 | C2 |....| Cn |
|------------------------|
| ZK |val1| ZK |....|time|
|------------------------|
| ZK |val2| ZK |....|time|
|------------------------|
|....|....| .. |....| ...|
|------------------------|
| ZK |valn| ZK |....|time|

|Ind | C1 | C2 |....| Cn |
|------------------------|
| QK |val1| QK |....|time|
|------------------------|
| QK |val2| QK |....|time|
|------------------------|
|....|....| .. |....| ...|
|------------------------|
| QK |valn| QK |....|time|

但是重新分区函数“合并”了最后两个索引，所以最后一个分区如下所示：

|Ind | C1 | C2 |....| Cn |
|------------------------|
| ZK |val1| ZK |....|time|
|------------------------|
| ZK |val2| ZK |....|time|
|------------------------|
|....|....| .. |....| ...|
|------------------------|
| QK |valn| QK |....|time|
|------------------------|
|....|....| .. |....| ...|
|------------------------|
| QK |valn| QK |....|time|

你知道为什么会这样吗？或者你有更好的办法来解决我的问题吗？我知道有一个dask.groupby(...).apply(...)。但是我的映射函数有副作用，apply(...)总是按设计为每个dask分区执行两次。在