Проблема при вычислении / объединении dask dataframe (s), когда индекс является категориальным - PullRequest
Новая
       20

Проблема при вычислении / объединении dask dataframe (s), когда индекс является категориальным

1 голос
/ 02 апреля 2020

Я пытаюсь использовать dask для обработки набора данных, который не помещается в память. Это данные временных рядов для различных «идентификаторов». После прочтения документации dask, я решил использовать формат файла "parquet" и разбиение по "ID".

Однако, читая из parquet и устанавливая индекс, я столкнулся с "TypeError: для объединения упорядоченных категорий, все категории должны быть одинаковыми », который мне не удалось решить самостоятельно.

Этот код повторяет проблему, с которой я столкнулся: 

import dask.dataframe as dd
import numpy as np
import pandas as pd
import traceback

# create ids
ids = ["AAA", "BBB", "CCC", "DDD"]

# create data
df = pd.DataFrame(index=np.random.choice(ids, 50), data=np.random.rand(50, 1), columns=["FOO"]).reset_index().rename(columns={"index": "ID"})
# serialize  to parquet
f = r"C:/temp/foo.pq"
df.to_parquet(f, compression='gzip', engine='fastparquet', partition_cols=["ID"])
# read with dask
df = dd.read_parquet(f)

try:
    df = df.set_index("ID")
except Exception as ee:
    print(traceback.format_exc())

, в этот момент я получаю следующую ошибку : 

~\.conda\envs\env_dask_py37\lib\site-packages\pandas\core\arrays\categorical.py in check_for_ordered(self, op)
   1492         if not self.ordered:
   1493             raise TypeError(
-> 1494                 f"Categorical is not ordered for operation {op}\n"
   1495                 "you can use .as_ordered() to change the "
   1496                 "Categorical to an ordered one\n"

TypeError: Categorical is not ordered for operation max
you can use .as_ordered() to change the Categorical to an ordered one

Затем я сделал: 

# we order the categorical
df.ID = df.ID.cat.as_ordered()
df = df.set_index("ID")

И когда я пытаюсь использовать df.compute(scheduler="processes"), я получаю ошибку TypeError, о которой я упоминал ранее: 

try:
    schd_str = 'processes'
    aa = df.compute(scheduler=schd_str)
    print(f"{schd_str}: OK")
except:
    print(f"{schd_str}: KO")
    print(traceback.format_exc())

дает: 

Traceback (most recent call last):
  File "<ipython-input-6-e15c4e86fee2>", line 3, in <module>
    aa = df.compute(scheduler=schd_str)
  File "C:\Users\xxx\.conda\envs\env_dask_py37\lib\site-packages\dask\base.py", line 166, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "C:\Users\xxx\.conda\envs\env_dask_py37\lib\site-packages\dask\base.py", line 438, in compute
    return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
  File "C:\Users\xxx\.conda\envs\env_dask_py37\lib\site-packages\dask\base.py", line 438, in <listcomp>
    return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
  File "C:\Users\xxx\.conda\envs\env_dask_py37\lib\site-packages\dask\dataframe\core.py", line 103, in finalize
    return _concat(results)
  File "C:\Users\xxx\.conda\envs\env_dask_py37\lib\site-packages\dask\dataframe\core.py", line 98, in _concat
    else methods.concat(args2, uniform=True, ignore_index=ignore_index)
  File "C:\Users\xxx\.conda\envs\env_dask_py37\lib\site-packages\dask\dataframe\methods.py", line 383, in concat
    ignore_index=ignore_index,
  File "C:\Users\xxx\.conda\envs\env_dask_py37\lib\site-packages\dask\dataframe\methods.py", line 431, in concat_pandas
    ind = concat([df.index for df in dfs])
  File "C:\Users\xxx\.conda\envs\env_dask_py37\lib\site-packages\dask\dataframe\methods.py", line 383, in concat
    ignore_index=ignore_index,
  File "C:\Users\xxx\.conda\envs\env_dask_py37\lib\site-packages\dask\dataframe\methods.py", line 400, in concat_pandas
    return pd.CategoricalIndex(union_categoricals(dfs), name=dfs[0].name)
  File "C:\Users\xxx\.conda\envs\env_dask_py37\lib\site-packages\pandas\core\dtypes\concat.py", line 352, in union_categoricals
    raise TypeError("Categorical.ordered must be the same")
TypeError: Categorical.ordered must be the same

Удивительно, но использование df.compute(scheduler="threads"), df.compute(scheduler="synchronous") или вообще отсутствие установки индекса работает правильно.

Однако мне кажется, что это не то, что я должен делать, поскольку я на самом деле пытаюсь объединить несколько из этих наборов данных и подумал, что установка индекса приведет к ускорению, а не к настройке. (Я получаю ту же самую ошибку, когда пытаюсь объединить два кадра данных, проиндексированных таким образом)

Я попытался проверить df._meta, и оказалось, что мои категории «известны», как и должно быть? dask-категорически

Я также прочитал этот пост на github о чем-то похожем, но каким-то образом не нашел решения.

Спасибо за вашу помощь

...