Dask dataframe - разбить столбец на несколько строк на основе разделителя

Question

Каков эффективный способ разделения столбца на несколько строк с использованием dask dataframe? Например, допустим, у меня есть файл csv, который я прочитал, используя dask, для создания следующего кадра данных dask:

id var1 var2
1  A    Z,Y
2  B    X
3  C    W,U,V

Я хотел бы преобразовать его в:

id var1 var2
1  A    Z
1  A    Y
2  B    X
3  C    W
3  C    U
3  C    V

Я изучил ответы для Разделить (взорвать) строку строки в панде данных панд, чтобы разделить строки и панд: Как разделить текст в столбце на несколько строк? .

Я пытался применить ответ, приведенный в https://stackoverflow.com/a/17116976/7275290, но dask, похоже, не принимает ключевое слово расширения в str.split.

Я также пытался применить векторизованный подход, предложенный в https://stackoverflow.com/a/40449726/7275290, но затем обнаружил, что np.repeat не реализован в dask с целочисленными массивами (https://github.com/dask/dask/issues/2946).

Я опробовал несколько других методов в пандах, но они были очень медленными - возможно, быстрее с dask, но я сначала хотел проверить, был ли кто-нибудь успешным с каким-либо конкретным методом. Я работаю с набором данных с более чем 10 миллионами строк и 10 столбцов (строковые данные). После разбиения на строки, это, вероятно, станет ~ 50 миллионов строк.

Спасибо, что заглянули в это! Я ценю его.

Answer 1

Используйте это:

>>> df.join(pd.DataFrame(df.var2.str.split(',', expand=True).stack()
            .reset_index(level=1, drop=True),columns=['var2 '])).drop('var2',1)
            .rename(columns=str.strip)
   id var1 var2
0   1    A    Z
0   1    A    Y
1   2    B    X
2   3    C    W
2   3    C    U
2   3    C    V
>>> 

Или, если необходимо сбросить индекс:

>>> df.join(pd.DataFrame(df.var2.str.split(',', expand=True).stack()                     
            .reset_index(level=1, drop=True),columns=['var2 '])).drop('var2',1)
            .rename(columns=str.strip).reset_index(drop=True)
   id var1 var2
0   1    A    Z
1   1    A    Y
2   2    B    X
3   3    C    W
4   3    C    U
5   3    C    V
>>> 

Для таблицы данных dask:

from dask import dataframe as dd 
sd = dd.from_pandas(df, npartitions=6)

Сроки (буквально одинаковые):

>>> timeit.timeit(lambda: df.join(pd.DataFrame(df.var2.str.split(',', expand=True).stack()
            .reset_index(level=1, drop=True),columns=['var2 '])).drop('var2',1)
            .rename(columns=str.strip),number=10) # U9-Forward
0.05815268672555618
>>> timeit.timeit(lambda: df.drop('var2', axis=1).join(
    df.var2.str.split(',', expand=True).stack().reset_index(drop=True, level=1).rename('var2')),number=10) # mdurant
0.05137591577754108
>>> 

Answer 2

Dask позволяет вам использовать панды непосредственно для операций, которые являются построчными (например, такими) или могут применяться по одному разделу за раз.Помните, что фрейм данных Dask состоит из набора фреймов данных Pandas.

Для случая Pandas вы сделаете это, основываясь на связанных вопросах:

df = pd.DataFrame([["A", "Z,Y"], ["B", "X"], ["C", "W,U,V"]], 
    columns=['var1', 'var2'])
df.drop('var2', axis=1).join(
    df.var2.str.split(',', expand=True).stack().reset_index(drop=True, level=1).rename('var2'))

, поэтому для Dask вы можете подать заявку в точноститот же метод через map_partitions, потому что каждая строка не зависит от всех других.Возможно, это выглядело бы чище, если бы переданная функция была записана отдельно, а не как лямбда:

d = dd.from_pandas(df, 2)
d.map_partitions(
    lambda df: df.drop('var2', axis=1).join(
        df.var2.str.split(',', expand=True).stack().reset_index(drop=True, level=1).rename('var2')))

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