<jupyter_start><jupyter_text>A la découverte des rdd On importe Spark et on lance un cluster en local. Grâce à "*", on lui indique qu'on veut autant de threads que de coeurs logiques sur la machine.https://stackoverflow.com/questions/32356143/what-does-local-mean-in-sparkAppelons la variable sc (pour Spark Context) Mon explication sur ce qu ej 'ai compris...<jupyter_code>import pyspark
sc = pyspark.SparkContext('local[*]')
a = 112
print(a)<jupyter_output>112<jupyter_text>Quelques informations sur le Spark Context<jupyter_code>sc.defaultParallelism
sc.defaultMinPartitions<jupyter_output><empty_output><jupyter_text>On crée notre premier rdd à partir de la fonction Python range.parallelize sert à indiquer à Spark qu'on désire lui faire créer un rdd.On peut créer un rdd à partir d'énormément de sources :- Un tableau d'entiers Python comme ici- Un tableau de dictionnaires Python- Un fichier csv- Un fichier json (séparé par des retours à la ligne)- Un fichier Avro...- Ces fichiers peuvent être sur HDFS, sur un filesystem classique (comme on va le faire), sur Amazon S3...Au moment où on fait parallelize, le rdd n'est pas encore véritablement créé.Il le sera quand on fera appel à takeSample qui prend un échantillon aléatoire de notre rdd à la taille demandée.takeSample est une action.<jupyter_code>rdd = sc.parallelize(range(1000))
rdd.takeSample(False, 5)<jupyter_output><empty_output><jupyter_text>Quand on en a fini avec le contexte Spark, on peut le clore. Ici il est en commentaire mais il faudra y penser plus tard. On le clôt en appelant la méthode stop du Spark Context<jupyter_code>sc.stop()<jupyter_output><empty_output><jupyter_text>Etudions un peu ce que renvoie range(1000) exactement<jupyter_code>[i for i in range(1000)]<jupyter_output><empty_output><jupyter_text>On peut maintenant écrire notre premier map<jupyter_code>rdd2 = rdd.map(lambda x: x + 1)<jupyter_output><empty_output><jupyter_text>collect est une nouvelle action (comme takeSample), sauf que takeSample prend un échantillon alors que collect rappatrie TOUT. Attention donc, à utiliser avec modération.Dans notre cas on aura pas de problème.On constate que l'opération consistant à rajouter 1 à tous les éléments s'est plutôt bien passée.On constate aussi qu'on a créé un nouvel rdd, rdd2.On rappelle que les rdd sont immutable. Je ne peux pas modifier l'objet rdd. Je peux évidemment affecter la variable rdd à autre chose, mais c'est une autre histoire.<jupyter_code>rdd2.collect()
Créons un troisième rdd
rdd3 = rdd.map(lambda x: x * 4)
rdd3.collect()<jupyter_output><empty_output><jupyter_text>Trouver un moyen de créer rdd4 qui contiendrait les éléments de rdd2 puis de rdd3<jupyter_code>#to be continued
rdd4 = rdd2.union(rdd3)
rdd5 = rdd2 + rdd3
rdd6 = rdd4.map(lambda x: str(x))

def printRdd(rdd):
    print(" ".join([str(x) for x in rdd.collect()]))

def printRddString(rdd):
    print(" ".join(rdd.collect()))

l1 = rdd2.collect()
l2 = rdd3.collect()
rdd4 = sc.parallelize(l1 + l2)

printRdd(rdd4)
printRddString(rdd6)<jupyter_output>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 [...]<jupyter_text>Trouver un moyen d'avoir le nombre d'éléments d'un rdd<jupyter_code>#nbElements = ...
print(rdd.count())
len(rdd.collect())<jupyter_output>1000<jupyter_text>Trouvez un moyen de récupérer (une action, pas une création d'un nouvel rdd) seulement le premier élément d'un rdd. Un tableau avec les premiers n éléments ?<jupyter_code># ...
print(rdd.first())
rdd.take(1)<jupyter_output>0<jupyter_text>inputArray est un tableau de dictionnaires Python. Vu sa taille, il mérite pleinement de devenir un rdd Spark. 1. Le transformer en rdd.2. collecter le rdd3. Constater que les valeurs se ressemblent4. Modifier inputArray. Le rdd est-il impacté ?5. Faire en sorte de créer un nouveau rdd avec pour chaque élément un nouvel élément nomComplet qui correspond à la concaténation du prénom et du nom, séparés évidemment par une virgule.6. Créer encore un nouvel rdd à partir de celui qu'on vient de créer. On voudra simplement le nombre de caractères du champ nomComplet. On veut donc un rdd de nombres.7. On veut encore un nouvel rdd avec seulement les éléments du rdd précédent valant au moins 128. filter est-il une action ou une transformation ?9. Quand on parle de transformation, entend-on modification du rdd en place ou création d'un nouvel rdd ?10. Comment garder que les valeurs distinctes sur le rdd qu'on avait en 6 (on veut un nouvel rdd)<jupyter_code>inputArray = [{"nom":"dore","prenom":"julien"},{"nom":"cesar","prenom":"jules"},{"nom":"foucault","prenom":"jean-pierre"},{"nom":"berger","prenom":"michel"}]
rdd = sc.parallelize(inputArray)
inputArray[0]["nom"] = "papy"
print(inputArray,"\n\n",rdd.collect())

def updateRow(x):
    x["nomComplet"] = x["prenom"] + " " + x["nom"]
    return x
rdd2 = rdd.map(updateRow)
print(rdd.collect(),"\n\n",rdd2.collect())
rdd3 = rdd2.map(lambda x: len(x["nomComplet"]))
printRdd(rdd3)

rdd4 = rdd3.filter(lambda x: x >= 12)
printRdd(rdd4)

printRdd(rdd3.distinct())<jupyter_output>[{'nom': 'papy', 'prenom': 'julien'}, {'nom': 'cesar', 'prenom': 'jules'}, {'nom': 'foucault', 'prenom': 'jean-pierre'}, {'nom': 'berger', 'prenom': 'michel'}] 

 [{'nom': 'dore', 'prenom': 'julien'}, {'nom': 'cesar', 'prenom': 'jules'}, {'nom': 'foucault', 'prenom': 'jean-pierre'}, {'nom': 'berger', 'prenom': 'michel'}]
[{'nom': 'dore', 'prenom': 'julien'}, {'nom': 'cesar', 'prenom': 'jules'}, {'nom': 'foucault', 'prenom': 'jean-pierre'}, {'nom': 'berger', 'prenom': 'michel'}] 

 [{'nom': 'dore', 'prenom': 'julien', 'nomComplet': '<NAME>'}, {'nom': 'cesar', 'prenom': 'jules', 'nomComplet': '<NAME>'}, {'nom': 'foucault', 'prenom': 'jean-pierre', 'nomComplet': '<NAME>'}, {'nom': 'berger', 'prenom': 'michel', 'nomComplet': '<NAME>'}]
11 11 20 13
20 13
20 13 11<jupyter_text>Les arbres de Paris d'hier au format CSV1. Trouver de quoi lire le fichier csv en mode texte classique2. Spark a-t-il effectué une transformation pour transformer les lignes du fichiers en "cellule" ?Nous allons effectuer une transformation basique et naïve.2. (bis) Combien y-a-t-il de lignes dans ce fichier ? De caractères valant ; ?3. Lire la première ligne du fichier et extraire les noms des colonnes4. Faire en sorte d'avoir un nouvel rdd avec non pas un tableau de lignes mais un tableau de dictionnaires indexés par les noms des colonnes.5. Spark (ou Python) considère-t-il une colonne comme des entiers ou des float ? Pourquoi ?6. Faire en sorte de traiter les entiers comme des entiers, les float comme des float (notamment hauteur...)<jupyter_code>rdd = sc.textFile("les-arbres.csv")
print("Nombre de lignes",rdd.count())
print(rdd.map(lambda ligne: ligne.count(";")).sum())

firstRow = rdd.first()
print("Première ligne",firstRow)
headers = firstRow.split(";")

def filterFirstRow(idx,iter):
    if idx == 0:
        next(iter)
    return iter

newRdd = rdd.mapPartitionsWithIndex(filterFirstRow)

#"JEROME;CHAABAN;18" ==> [“JEROME,“CHAABAN”,“18"] ==> {nom:CHAABAN,prenom:JEROME,age:18}

def fromLineToDict(line):
    splittedLine = line.split(";")
    toBeReturn = {header:splittedLine[headerNumber] for headerNumber,header in enumerate(headers)}
    toBeReturn["HAUTEUR (m)"] = float(toBeReturn["HAUTEUR (m)"])
    return toBeReturn

rddDict = newRdd.map(fromLineToDict)
print(rddDict.take(3))

print(rddDict.filter(lambda x: x["HAUTEUR (m)"] >= 10).count())
import time
#Création d'une fonction classique Python
def maFonction():
    return 112
def maFonction2(x):
    time.sleep( 5 )
    return 4 + x
#print(maFonction())
#print(maFonction2(4))
myArray = [5,4,2,6,10]
rdd = sc.parallelize(myArray)
#rdd2 = rdd.map(lambda x: 4 + x)
rdd3 = rdd.map(maFonction2)

print("Start : %s" % time.ctime())
rdd3.collect()
print("Stop : %s" % time.ctime())
[maFonction2(x) for x in myArray]
print("Stop 2 : %s" % time.ctime())
#rdd3.collect()
a = 58
b = a
a = 89

a = {"prenom":"Jerome","nom":"Chaaban"}
b = a
a["prenom"] = "Julien"

print(a,b)

a = [5,10]
b = a
a.append(56)
print(a,b)

a = ["hhh","kikkk"]
b = [5,10]
a.append(56)
a.append([76,5674])
print(a,b)

def manipulerDico(dico):
    dico["blabla"] = 56
    #return dico

a = {"prenom":"jerome","nom":"chaaban"}
manipulerDico(a)
print(a)

def manipulerTuple(aTuple):
    aTuple[3].append(545)
    #return dico
    
a = (54,67,65,[544,676])
print(a)
a[3].append(565)
manipulerTuple(a)
print(a)

def generateSuperiorFilterFunction(nb):
    def myFilter(x):
        return x > nb
    return myFilter
myArray = [4,3,10,10,10,10,54]
rdd = sc.parallelize(myArray)
print(rdd.filter(lambda x: x > 9).collect())
print(rdd.filter(lambda x: x > 2).collect())
print(rdd.filter(lambda x: x > 40).collect())<jupyter_output>[10, 10, 10, 10, 54]
[4, 3, 10, 10, 10, 10, 54]
[54]
{'fraction_non_alphanumeric': 0.12520644095788605, 'fraction_numerical': 0.09805945499587118, 'mean_word_length': 4.9661330049261085, 'pattern_counts': {'":': 12, '<': 47, '<?xml version=': 0, '>': 55, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 21, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '401505', 'license': 'MIT', 'n_tokens_mistral': 4456, 'n_tokens_neox': 3591, 'n_words': 1250, 'path': 'spark/docker-spark/data/Formation.ipynb', 'repo_name': 'radioactivit/big-data', 'size': '9719'}