Dask dataframe¶

Un dataframe est une structure de données qui permet de stockers données tabulaires (lignes et colonnes où chaque colonne est identifiée par un nom). Il est similaire à une table dans une base de données ou à une feuille de calcul dans Excel.

Les dataframes Dask sont conçus pour traiter de gros

In [ ]:

import numpy as np
import pandas as pd

import dask.dataframe as dd

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

Création de 2400 dates séaprées de 1h

In [ ]:

index = pd.date_range("2021-09-01", periods=2400, freq="1h")
index

index = pd.date_range("2021-09-01", periods=2400, freq="1h") index

Out[ ]:

DatetimeIndex(['2021-09-01 00:00:00', '2021-09-01 01:00:00',
               '2021-09-01 02:00:00', '2021-09-01 03:00:00',
               '2021-09-01 04:00:00', '2021-09-01 05:00:00',
               '2021-09-01 06:00:00', '2021-09-01 07:00:00',
               '2021-09-01 08:00:00', '2021-09-01 09:00:00',
               ...
               '2021-12-09 14:00:00', '2021-12-09 15:00:00',
               '2021-12-09 16:00:00', '2021-12-09 17:00:00',
               '2021-12-09 18:00:00', '2021-12-09 19:00:00',
               '2021-12-09 20:00:00', '2021-12-09 21:00:00',
               '2021-12-09 22:00:00', '2021-12-09 23:00:00'],
              dtype='datetime64[ns]', length=2400, freq='h')

Création d'un dataframe pandas de dates avec 3 colonnes.

1. "a" : un indetifiant de 0 à 2399
2. "b" : un caratère pris de la liste "abcaddbe" * 300 (8 caractères * 300 = 2400 lignes) un par un et répartis sur chaque ligne
3. la colonne index : une date parmi les 2400 dates générées plus haut (peremt de labéliser ou identifier la ligne de façon unique, comme la clé primaire dans une BDD relationnelle)

In [ ]:

df = pd.DataFrame({"a": np.arange(2400), "b": list("abcaddbe" * 300)}, index=index)
df

df = pd.DataFrame({"a": np.arange(2400), "b": list("abcaddbe" * 300)}, index=index) df

Out[ ]:

	a	b
2021-09-01 00:00:00	0	a
2021-09-01 01:00:00	1	b
2021-09-01 02:00:00	2	c
2021-09-01 03:00:00	3	a
2021-09-01 04:00:00	4	d
...	...	...
2021-12-09 19:00:00	2395	a
2021-12-09 20:00:00	2396	d
2021-12-09 21:00:00	2397	d
2021-12-09 22:00:00	2398	b
2021-12-09 23:00:00	2399	e

2400 rows × 2 columns

Création d'un dataframe dask à partir du dataframe pandas en demandant de répartir les données sur 10 paprtitions.

In [ ]:

ddf = dd.from_pandas(df, npartitions=10)
ddf

ddf = dd.from_pandas(df, npartitions=10) ddf

Out[ ]:

Dask DataFrame Structure:

	a	b
npartitions=10
2021-09-01 00:00:00	int64	string
2021-09-11 00:00:00	...	...
...	...	...
2021-11-30 00:00:00	...	...
2021-12-09 23:00:00	...	...

Dask Name: frompandas, 1 expression

Vérification des indicdes de début et de fin de chaque partition (l'indice de fin d'une partition est l'indice de début de la partition suivante - 1).

In [ ]:

ddf.divisions

ddf.divisions

Out[ ]:

(Timestamp('2021-09-01 00:00:00'),
 Timestamp('2021-09-11 00:00:00'),
 Timestamp('2021-09-21 00:00:00'),
 Timestamp('2021-10-01 00:00:00'),
 Timestamp('2021-10-11 00:00:00'),
 Timestamp('2021-10-21 00:00:00'),
 Timestamp('2021-10-31 00:00:00'),
 Timestamp('2021-11-10 00:00:00'),
 Timestamp('2021-11-20 00:00:00'),
 Timestamp('2021-11-30 00:00:00'),
 Timestamp('2021-12-09 23:00:00'))

Accès à une partition particulière

In [ ]:

ddf.partitions[1]

ddf.partitions[1]

Out[ ]:

Dask DataFrame Structure:

	a	b
npartitions=1
2021-09-11	int64	string
2021-09-21	...	...

Dask Name: partitions, 2 expressions

Indexation¶

API simiaire à celle du DataFrame pandas et de des tableaus NumPy.

L'exemple suivant sélectionne la colonne "b" avec le DataFrame dask et le DataFrame pandas.

In [ ]:

df["b"] # Ou df.b

df["b"] # Ou df.b

Out[ ]:

2021-09-01 00:00:00    a
2021-09-01 01:00:00    b
2021-09-01 02:00:00    c
2021-09-01 03:00:00    a
2021-09-01 04:00:00    d
                      ..
2021-12-09 19:00:00    a
2021-12-09 20:00:00    d
2021-12-09 21:00:00    d
2021-12-09 22:00:00    b
2021-12-09 23:00:00    e
Freq: h, Name: b, Length: 2400, dtype: object

In [ ]:

ddf.b # ou ddf["a"]

ddf.b # ou ddf["a"]

Out[ ]:

Dask Series Structure:
npartitions=10
2021-09-01 00:00:00    string
2021-09-11 00:00:00       ...
                        ...  
2021-11-30 00:00:00       ...
2021-12-09 23:00:00       ...
Dask Name: getitem, 2 expressions
Expr=df['b']

Sélection d'un intervalles d'indices

In [ ]:

df["2021-10-01": "2021-10-09 5:00"]

df["2021-10-01": "2021-10-09 5:00"]

Out[ ]:

	a	b
2021-10-01 00:00:00	720	a
2021-10-01 01:00:00	721	b
2021-10-01 02:00:00	722	c
2021-10-01 03:00:00	723	a
2021-10-01 04:00:00	724	d
...	...	...
2021-10-09 01:00:00	913	b
2021-10-09 02:00:00	914	c
2021-10-09 03:00:00	915	a
2021-10-09 04:00:00	916	d
2021-10-09 05:00:00	917	d

198 rows × 2 columns

In [ ]:

ddf["2021-10-01": "2021-10-09 5:00"]

ddf["2021-10-01": "2021-10-09 5:00"]

Out[ ]:

Dask DataFrame Structure:

	a	b
npartitions=1
2021-10-01 00:00:00.000000000	int64	string
2021-10-09 05:00:59.999999999	...	...

Dask Name: loc, 2 expressions

On remarque que dask n'évalue pas tout de suite l'opération (lazy).

Calcul ou évaluation des résultats¶

Dask est paresseux (lazy). Il faut donc appeler la méthode compute() pour évaluer les résultats.

In [ ]:

ddf["2021-10-01": "2021-10-09 5:00"].compute()

ddf["2021-10-01": "2021-10-09 5:00"].compute()

Out[ ]:

	a	b
2021-10-01 00:00:00	720	a
2021-10-01 01:00:00	721	b
2021-10-01 02:00:00	722	c
2021-10-01 03:00:00	723	a
2021-10-01 04:00:00	724	d
...	...	...
2021-10-09 01:00:00	913	b
2021-10-09 02:00:00	914	c
2021-10-09 03:00:00	915	a
2021-10-09 04:00:00	916	d
2021-10-09 05:00:00	917	d

198 rows × 2 columns

Méthodes (ou api)¶

Simlaire à NumPy et pandas, sauf qu'il faut considérer que Dask est paresseux (lazy) et qu'il faut appeler la méthode compute() pour évaluer les résultats.

In [ ]:

ddf.head()

ddf.head()

Out[ ]:

	a	b
2021-09-01 00:00:00	0	a
2021-09-01 01:00:00	1	b
2021-09-01 02:00:00	2	c
2021-09-01 03:00:00	3	a
2021-09-01 04:00:00	4	d

In [ ]:

ddf.a.mean() # Non ca

ddf.a.mean() # Non ca

Out[ ]:

<dask_expr.expr.Scalar: expr=df['a'].mean(), dtype=float64>

In [ ]:

ddf.a.mean().compute()

ddf.a.mean().compute()

Out[ ]:

np.float64(1199.5)

In [ ]:

ddf.b.unique()

ddf.b.unique()

Out[ ]:

Dask Series Structure:
npartitions=10
    string
       ...
     ...  
       ...
       ...
Dask Name: unique, 3 expressions
Expr=Unique(frame=df['b'])

In [ ]:

ddf.b.unique().compute()

ddf.b.unique().compute()

Out[ ]:

0    e
0    c
1    d
0    a
0    b
Name: b, dtype: string

On peut chainer les appels de méthodes.

Ici, on récupère la somme cumulée de la colonne "a" pour l'intervalle ["2021-10-01": "2021-10-09 5:00"] et on lui soustrait 100.

In [ ]:

result = ddf["2021-10-01": "2021-10-09 5:00"].a.cumsum() - 100
result

result = ddf["2021-10-01": "2021-10-09 5:00"].a.cumsum() - 100 result

Out[ ]:

Dask Series Structure:
npartitions=1
2021-10-01 00:00:00.000000000    int64
2021-10-09 05:00:59.999999999      ...
Dask Name: sub, 5 expressions
Expr=CumSum(frame=(LocSlice(frame=df, iindexer=slice(Timestamp('2021-10-01 00:00:00'), Timestamp('2021-10-09 05:00:59.999999999'), None)))['a'], skipna=True) - 100

In [ ]:

result.compute()

result.compute()

Out[ ]:

2021-10-01 00:00:00       620
2021-10-01 01:00:00      1341
2021-10-01 02:00:00      2063
2021-10-01 03:00:00      2786
2021-10-01 04:00:00      3510
                        ...  
2021-10-09 01:00:00    158301
2021-10-09 02:00:00    159215
2021-10-09 03:00:00    160130
2021-10-09 04:00:00    161046
2021-10-09 05:00:00    161963
Freq: h, Name: a, Length: 198, dtype: int64

Somme cumulée¶

Calcule la somme au fut et à mesure des lignes. La somme totale ou final se situera à la dernière ligne.

Voici un exemple avec Pandas (Series et une DataFrame en une seule dimension, ou une colonne d'un DataFrame).

In [ ]:

s = pd.Series([2, np.nan, 5, -1, 0])
s

s = pd.Series([2, np.nan, 5, -1, 0]) s

Out[ ]:

0    2.0
1    NaN
2    5.0
3   -1.0
4    0.0
dtype: float64

In [ ]:

s.cumsum()

s.cumsum()

Out[ ]:

0    2.0
1    NaN
2    7.0
3    6.0
4    6.0
dtype: float64

Graphe de tâches¶

Grâhe de tâches de la variable result qui correspond à la somme cumulée calculée précédemment. La méthode visualize() nécessite d'installer le binaire de graphviz (via votre gestionnaire de paquets ou via un installeur fourni sur le site officiel) ainsiq que le module qui porte le même nom.

In [ ]:

result.dask

result.dask

Out[ ]:

{('sub-5a8ce658ce881b9a168109251a3109de',
  0): <Task ('sub-5a8ce658ce881b9a168109251a3109de', 0) sub(...)>,
 ('cumulativefinalize-26a6f68b010d9b3b81956d1fe88e1d08',
  0): ('cumsum-dc13e540bc07335397aa375278c5fa3f', 0),
 ('operation-5630c696a59039a7a9170475b5e6abe8',
  0): <Task ('operation-5630c696a59039a7a9170475b5e6abe8', 0) operation(...)>,
 ('cumsum-dc13e540bc07335397aa375278c5fa3f',
  0): <Task ('cumsum-dc13e540bc07335397aa375278c5fa3f', 0) cumsum(...)>,
 ('getitem-f1025e3e21b7f218b241648926d54bdc',
  0): <Task ('getitem-f1025e3e21b7f218b241648926d54bdc', 0) getitem(...)>,
 ('loc-fff6b95ce0f38ef749afb35f08fb5c77',
  0): <Task ('loc-fff6b95ce0f38ef749afb35f08fb5c77', 0) loc(...)>,
 ('partitions-bf0af511f3f618def961c7abe2804060',
  0): Alias(('partitions-bf0af511f3f618def961c7abe2804060', 0)->('frompandas-9e3ad60c306d3e285fbbed807f12dcbe', 3)),
 ('frompandas-9e3ad60c306d3e285fbbed807f12dcbe',
  0): DataNode(                       a  b
 2021-09-01 00:00:00    0  a
 2021-09-01 01:00:00    1  b
 2021-09-01 02:00:00    2  c
 2021-09-01 03:00:00    3  a
 2021-09-01 04:00:00    4  d
 ...                  ... ..
 2021-09-10 19:00:00  235  a
 2021-09-10 20:00:00  236  d
 2021-09-10 21:00:00  237  d
 2021-09-10 22:00:00  238  b
 2021-09-10 23:00:00  239  e
 
 [240 rows x 2 columns]),
 ('frompandas-9e3ad60c306d3e285fbbed807f12dcbe',
  1): DataNode(                       a  b
 2021-09-11 00:00:00  240  a
 2021-09-11 01:00:00  241  b
 2021-09-11 02:00:00  242  c
 2021-09-11 03:00:00  243  a
 2021-09-11 04:00:00  244  d
 ...                  ... ..
 2021-09-20 19:00:00  475  a
 2021-09-20 20:00:00  476  d
 2021-09-20 21:00:00  477  d
 2021-09-20 22:00:00  478  b
 2021-09-20 23:00:00  479  e
 
 [240 rows x 2 columns]),
 ('frompandas-9e3ad60c306d3e285fbbed807f12dcbe',
  2): DataNode(                       a  b
 2021-09-21 00:00:00  480  a
 2021-09-21 01:00:00  481  b
 2021-09-21 02:00:00  482  c
 2021-09-21 03:00:00  483  a
 2021-09-21 04:00:00  484  d
 ...                  ... ..
 2021-09-30 19:00:00  715  a
 2021-09-30 20:00:00  716  d
 2021-09-30 21:00:00  717  d
 2021-09-30 22:00:00  718  b
 2021-09-30 23:00:00  719  e
 
 [240 rows x 2 columns]),
 ('frompandas-9e3ad60c306d3e285fbbed807f12dcbe',
  3): DataNode(                       a  b
 2021-10-01 00:00:00  720  a
 2021-10-01 01:00:00  721  b
 2021-10-01 02:00:00  722  c
 2021-10-01 03:00:00  723  a
 2021-10-01 04:00:00  724  d
 ...                  ... ..
 2021-10-10 19:00:00  955  a
 2021-10-10 20:00:00  956  d
 2021-10-10 21:00:00  957  d
 2021-10-10 22:00:00  958  b
 2021-10-10 23:00:00  959  e
 
 [240 rows x 2 columns]),
 ('frompandas-9e3ad60c306d3e285fbbed807f12dcbe',
  4): DataNode(                        a  b
 2021-10-11 00:00:00   960  a
 2021-10-11 01:00:00   961  b
 2021-10-11 02:00:00   962  c
 2021-10-11 03:00:00   963  a
 2021-10-11 04:00:00   964  d
 ...                   ... ..
 2021-10-20 19:00:00  1195  a
 2021-10-20 20:00:00  1196  d
 2021-10-20 21:00:00  1197  d
 2021-10-20 22:00:00  1198  b
 2021-10-20 23:00:00  1199  e
 
 [240 rows x 2 columns]),
 ('frompandas-9e3ad60c306d3e285fbbed807f12dcbe',
  5): DataNode(                        a  b
 2021-10-21 00:00:00  1200  a
 2021-10-21 01:00:00  1201  b
 2021-10-21 02:00:00  1202  c
 2021-10-21 03:00:00  1203  a
 2021-10-21 04:00:00  1204  d
 ...                   ... ..
 2021-10-30 19:00:00  1435  a
 2021-10-30 20:00:00  1436  d
 2021-10-30 21:00:00  1437  d
 2021-10-30 22:00:00  1438  b
 2021-10-30 23:00:00  1439  e
 
 [240 rows x 2 columns]),
 ('frompandas-9e3ad60c306d3e285fbbed807f12dcbe',
  6): DataNode(                        a  b
 2021-10-31 00:00:00  1440  a
 2021-10-31 01:00:00  1441  b
 2021-10-31 02:00:00  1442  c
 2021-10-31 03:00:00  1443  a
 2021-10-31 04:00:00  1444  d
 ...                   ... ..
 2021-11-09 19:00:00  1675  a
 2021-11-09 20:00:00  1676  d
 2021-11-09 21:00:00  1677  d
 2021-11-09 22:00:00  1678  b
 2021-11-09 23:00:00  1679  e
 
 [240 rows x 2 columns]),
 ('frompandas-9e3ad60c306d3e285fbbed807f12dcbe',
  7): DataNode(                        a  b
 2021-11-10 00:00:00  1680  a
 2021-11-10 01:00:00  1681  b
 2021-11-10 02:00:00  1682  c
 2021-11-10 03:00:00  1683  a
 2021-11-10 04:00:00  1684  d
 ...                   ... ..
 2021-11-19 19:00:00  1915  a
 2021-11-19 20:00:00  1916  d
 2021-11-19 21:00:00  1917  d
 2021-11-19 22:00:00  1918  b
 2021-11-19 23:00:00  1919  e
 
 [240 rows x 2 columns]),
 ('frompandas-9e3ad60c306d3e285fbbed807f12dcbe',
  8): DataNode(                        a  b
 2021-11-20 00:00:00  1920  a
 2021-11-20 01:00:00  1921  b
 2021-11-20 02:00:00  1922  c
 2021-11-20 03:00:00  1923  a
 2021-11-20 04:00:00  1924  d
 ...                   ... ..
 2021-11-29 19:00:00  2155  a
 2021-11-29 20:00:00  2156  d
 2021-11-29 21:00:00  2157  d
 2021-11-29 22:00:00  2158  b
 2021-11-29 23:00:00  2159  e
 
 [240 rows x 2 columns]),
 ('frompandas-9e3ad60c306d3e285fbbed807f12dcbe',
  9): DataNode(                        a  b
 2021-11-30 00:00:00  2160  a
 2021-11-30 01:00:00  2161  b
 2021-11-30 02:00:00  2162  c
 2021-11-30 03:00:00  2163  a
 2021-11-30 04:00:00  2164  d
 ...                   ... ..
 2021-12-09 19:00:00  2395  a
 2021-12-09 20:00:00  2396  d
 2021-12-09 21:00:00  2397  d
 2021-12-09 22:00:00  2398  b
 2021-12-09 23:00:00  2399  e
 
 [240 rows x 2 columns])}

In [ ]:

result.visualize()

result.visualize()

Out[ ]:

In [ ]:

uq = ddf.b.unique()
uq.compute()
uq.visualize()

uq = ddf.b.unique() uq.compute() uq.visualize()

Out[ ]: