<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 2A.i - Données non structurées, programmation fonctionnelle 
Step1:  Données 
Step2:  On extrait les données de la base et on les écrit dans un fichiers plat.
Step3:  On sépare ce fichier plat en 50 morceaux.
Step4:  On crée un objet de type dask.bag pour gérer l'ensemble de ces fichiers.
Step5:  On va ensuite pouvoir utiliser la syntaxe cytoolz sur l'objet dbag.
dask va alors automatiquement gérer la parallélisation sur les différents fichiers.
Step6:  Attention à l'état de votre mémoire quand vous utilisez dask.
Step7:  De la même façon, dask vous permet de paralléliser efficacement des opérations effectuées avec numpy.
Step8:  Si vous avez des données obtenues de façon classique sur numpy (ici générées de façon aléatoires)
Step9:  dask gère de façon transparente cet ensemble de matrice comme une seule matrice de dimension supérieure. Ci-dessous x est une matrice de taille 100 X 5000 X 5000, ce qui ne tiendrait absolument pas en mémoire.
<ASSISTANT_TASK:>
Python Code:
from jyquickhelper import add_notebook_menu
add_notebook_menu()
Explanation: 2A.i - Données non structurées, programmation fonctionnelle : dask
dask est une sorte de mélange entre pandas et map/reduce. Le module implémente un sous-ensemble des possibilités de pandas sur des données qui ne tiennent pas nécessairement en mémoire.
End of explanation
import pyensae.datasource
pyensae.datasource.download_data("twitter_for_network_100000.db.zip")
import cytoolz as ct # import groupby, valmap, compose
import cytoolz.curried as ctc ## pipe, map, filter, get
import sqlite3
import pprint
try:
    import ujson as json
except:
    import json
conn_sqlite = sqlite3.connect("twitter_for_network_100000.db")
cursor_sqlite = conn_sqlite.cursor()
import os, psutil, gc, sys
if sys.platform.startswith("win"):
    pass
else:
    # only on Linux
    import resource
def memory_usage_psutil():
    gc.collect()
    process = psutil.Process(os.getpid())
    mem = process.memory_info()[0] / float(2 ** 20)
    print( "Memory used : %i MB" % mem )
    if not sys.platform.startswith("win"):        
        print( "Max memory usage : %i MB" % (resource.getrusage(resource.RUSAGE_SELF).ru_maxrss//1024) )
Explanation: Données : twitter_for_network_100000.db.zip or twitter_for_network_100000.db.zip (xavierdupre.fr).
Troisième partie : dask
Dask se présente comme une surcouche à toolz/cytoolz et numpy. Il n'offre pas de nouvelles fonctionnalités, mais vous permet de paralléliser les fonctions existantes.
End of explanation
cursor_sqlite.execute("select content FROM tw_users")
with open("tw_users_all.json", 'w') as f:
    ct.count( ct.map( f.write, ct.interpose( "\n", ct.pluck(0, cursor_sqlite ) ) ) )
Explanation: On extrait les données de la base et on les écrit dans un fichiers plat.
End of explanation
from tkinterquickhelper.funcwin.default_functions import file_split
file_split("tw_users_all.json", 50 )
Explanation: On sépare ce fichier plat en 50 morceaux.
End of explanation
import dask.bag as dbag
try:
    import ujson as json
except:
    import json
from operator import add
a = dbag.read_text('tw_users_all.json.split.*.json')
Explanation: On crée un objet de type dask.bag pour gérer l'ensemble de ces fichiers.
End of explanation
b = a.map(json.loads).pluck("followers_count").fold(add).compute()
%timeit -n1 b = a.map(json.loads).pluck("followers_count").fold(add).compute()
b
Explanation: On va ensuite pouvoir utiliser la syntaxe cytoolz sur l'objet dbag.
dask va alors automatiquement gérer la parallélisation sur les différents fichiers.
End of explanation
useless = [it**2 for it in range(25000000)]
#del useless
Explanation: Attention à l'état de votre mémoire quand vous utilisez dask.
End of explanation
import numpy
import dask.array as da
big_random = da.random.normal( 1000, 20000, size = (50000,50000), chunks = (1000,1000) )
big_random.mean().compute()
Explanation: De la même façon, dask vous permet de paralléliser efficacement des opérations effectuées avec numpy.
End of explanation
import numpy as np
import h5py
for it in range(10):
    a = np.random.random(size=(5000,5000))
    h5f = h5py.File('data_{0:02d}.h5'.format(it), 'w')
    h5f.create_dataset('dataset_1', data=a)
    h5f.close()
    if it % 5 == 0:
        print(it)
Explanation: Si vous avez des données obtenues de façon classique sur numpy (ici générées de façon aléatoires)
End of explanation
from glob import glob
import h5py
import dask.array as da
files = list(sorted(glob('data_*.h5'))) * 10
dsets = [h5py.File(fn)['dataset_1'] for fn in files]
arrays = [da.from_array(dset, chunks=(1000, 1000)) for dset in dsets]
x = da.stack(arrays, axis=0)
print(x.shape)
print( x[:,0,0].sum().compute() )
print( x[:,:,:].sum().compute() )
memory_usage_psutil()
Explanation: dask gère de façon transparente cet ensemble de matrice comme une seule matrice de dimension supérieure. Ci-dessous x est une matrice de taille 100 X 5000 X 5000, ce qui ne tiendrait absolument pas en mémoire.
End of explanation
<END_TASK>