MR - Spark

Objectif Découvrir Apache Spark avec Scala

Installation

Télécharger une version 2.x (2.1 ou 2.2 en octobre 2017) de Spark sur le site spark.apache.org.

Manipulation

On va manipuler Spark à travers son shell, dans le répertoire de Spark : cd bin/ puis ./spark-shell

Aide Dans le shell Spark, vous pouvez avoir de l'aide contextuel en tapant une tabulation après l'opérateur '.'. Vous pouvez avoir la documentation spark à l'URL http://spark.apache.org/

Vous pouvez accéder à une structure composée dans une opération Spark à l'aide de case.

Exemple pour une RDD rdd1 du type tuple de 2 éléments : rdd1.map({case(a,b)=> ..})

Exercice

On va charger le fichier du Alice aux pays des merveilles (télécharger depuis le lien de la page du

cours) val text = sc.textFile(''/mon/repertoire/aliceInWonderland'') On va commencer par regarder comment le RDD se présente text.collect ou bien text.take(n) pour avoir les n premières lignes. On voit qu'il y a entrée du tableau par ligne du fichier. On commence par compter le nombre de lignes du fichier à l'aide de l'action count.

1. text.count

2. Supprimer les lignes dont la longueur est nulle (length). On va utiliser la transformation Spark

filter. Compter le nombre de lignes.

3. A l'aide de la fonction Scala contains, on veut compter le nombre de 'CHAPTER' de l'ouvrage

4. On veut créer une nouvelle RDD qui va contenir le nombre de caractères par ligne. Donner la

valeur de la ligne contenant le plus grand nombre de caractères. Vous devez utiliser l'action reduce.

5. On veut écrire un wordCount en Spark sur le texte. A partir du résultat de la question 2, on veut

compter le nombre d'occurrence de chaque mot dans le texte.

5.a On veut créer une entrée dans une nouvelle RDD par mot, c'est-à-dire, qu'il faut

découper les lignes par occurrences de '' ''. Pour cela, vous pouvez utiliser la transformation flatMap et la fonction split('' '').

5b. A partir de la RDD obtenue en 5a, écrire le wordCount en vous inspirant de l'approche

MapReduce vu en cours. Combien d'entrées avez-vous dans le résultat.

6. On veut obtenir une liste ordonnée du résultat obtenu en question 5

6a. Ordonnée par mot (ordre alphabétique). Fournir le premier mot et le nombre

d'occurrence.

6b. Ordonnée par nombre décroissant d'occurrences. Fournir le mot apparaissant le plus

grand nombre de fois et son nombre d'occurrence.

7. Nous n'avons pas nettoyé le texte avant le traitement, cela explique qu'il y a plusieurs entrées

dans la RDD résultante de 6a avec le mot "Alice»

7a. Vous devez créer une RDD ne contenant que les entrées avec le terme ''Alice''.

Combien en trouvez vous.

7b. Faire le cumul des valeurs du RDD obtenue en 7a.

8. Nous allons changer de jeu de données pour étudier les possibilités de jointure avec Spark.

8a. Nous allons charger les jeux de données CIP_GENER_bdpm.txt et CIS_CPD_bdpm.txt

dans deux RDD distinctes que l'on nomme respectivement gen et cpd dans la suite. Fournir les tailles des 2 RDDs.

8b.Dans les 2 fichiers, les séparateurs de champs sont '\t'. Pour le fichier gen, on ne

s'intéresse qu'aux champs 3 (code CIS) et 4 (type générique), e.g., 61872344 et 1 sur la première

ligne. Pour le second fichier, nous nous intéressons à l'ensemble des champs (code CIS et intitulé

liste). Modifier les RDD pour ne garder que les champs nous intéressant.

8c. Nous voulons obtenir la liste des codes CIS et de l'intitulé de la liste pour les

médicaments ayant un type générique égal à 0. Cela correspond donc à une jointure entre 2

RDDs. Donner le nombre de lignes du résultat.

8d. Avons-nous des doublons de codes CIS dans la RDD résultante de 8c ?

8e. Combien y-a-t-il de doublons ?

9. On considère le RDD suivant :

var triples = sc.parallelize(Array((1,0,5),(5,1,8),(8,2,1),(2,0,6),(3,0,6),(6,1,9),(5,1,9),(9,3,11), (9,4,12),(4,0,7),(7,1,9),(7,2,10)))

Les valeurs des triples correspondent à l'encodage de respectivement un sujet, un prédicat et un

objet d'un graphe (que l'on considère comme orienté).

9a. Créer un RDD ne contenant que les sujets et objets du graphe.

9b. Trouver les noeuds du graphe qui sont des racines (pas d'arc entrant sur ces noeuds).

9c. Créer un RDD qui contiendra l'ensemble des noeuds accessibles depuis chaque noeud

(fermeture transitive des arcs pour chaque noeud). Exemple : a → b → c, nous avions déjà (a,b),

(b,c), il faut ajouter (a,c).

9d. A partir de ce dernier RDD et du RDD contenant les racines, former un RDD (rootedRSG) qui

contiendra l'ensemble des chemins de chaque racine.

Solution

1. text.count

// 3339

2. val text2 = text.filter(x=>x.length()>0)

text2.count // 2481

3. val chap = text.filter(x=>x.contains("CHAPTER"))

chap.count //12

4. val l = text2.map(x=>x.length()).reduce((a,b)=>a.max(b))

5a. val words = text2.flatMap(x=>x.split(" "))

5b. val wordCount = words.map(x=>(x,1)).reduceByKey(_+₎ wordCount.count // 5299

6a. val wcSorted = wordCount.sortByKey()

// ("",1081)

6b. val wcs2 = wordCount.map({case(w,n)=>(n,w)}).sortByKey()

wcs2.top(10) // (1507,the)

7a. val alice = wcs2.filter({case(n,w)=>w.contains("Alice")})

alice.count // 13

7b. val aliceCount = alice.map({case(n,w)=>n})

aliceCount.reduce(_+_).count // 395

8a. val cpd = sc.textFile("/home/oliv/Documents/CIS_CPD_bdpm.txt")

// 17523 val gen = sc.textFile("/home/oliv/Documents/CIS_GENER_bdpm.txt") // 8472

8b. val gen2 = gen.map(x=>x.split("\t")).map(t=>(t(2),t(3)))

val cpd2 = cpd.map(x=>x.split("\t")).map(t=>(t(0),t(1)))

8c. val cisList = gen2.filter({case(c,g)=>g.toLong==0}).join(cpd2)

cisList.count //1528

8d. val cis = cisList.map({case(c,(t,l))=> c}).distinct.count

// 930 oui, il y a des doublons

8e. val doubl = cis.map(x=>(x,1)).reduceByKey(_+_).filter{case(c,n)=>n>1}

doubl.count // res51: Long = 209 9. import scala.collection.mutable.ListBuffer import org.apache.spark.HashPartitioner import org.apache.spark.mllib.clustering.KMeans import org.apache.spark.mllib.linalg.Vectors val folderName = folder +scale val fileName = dataset+scale+"_encoded_unique_quads.part."+part+".2hop" val NB_FRAGMENTS = 300 // max sc.defaultParallelism var storageLevel = org.apache.spark.storage.StorageLevel.MEMORY_ONLY var start = java.lang.System.currentTimeMillis(); var triples = sc.parallelize(Array((1,0,5),(5,1,8),(8,2,1),(2,0,6),(3,0,6),(6,1,9),(5,1,9),(9,3,11), (9,4,12),(4,0,7),(7,1,9),(7,2,10))) var soPairs = triples.map({case(s,p,o)=>(s,o)})

var soPairsT = sc.parallelize(Array((1,5),(5,8),(8,1),(2,6),(3,6),(6,9),(5,9),(9,11),(9,12),(4,7),(7,9),

(7,10)))

var soPairs = sc.parallelize(Array((1,5),(5,8),(8,1),(2,6),(3,6),(6,9),(5,9),(9,11),(4,7),(7,9),(7,10),

(12,13),(13,10),(13,14),(14,12))) val subjects = soPairs.map({case(s,o)=>(s)}).distinct val objects = soPairs.map({case(s,o)=>(o)}).distinct subjects.persist(storageLevel) var roots = subjects.subtract(soPairs.map({case(s,o)=>(o)})).map(x=>(x,x)).distinct var leaves = objects.subtract(subjects) val edges = soPairs.map(x => (x._2, x._1)) var oldCount = 0L var nextCount = soPairs.count() do { oldCount = nextCount // Perform the join, obtaining an RDD of (y, (z, x)) pairs, // then project the result to obtain the new (x, z) paths. soPairs = soPairs.union(soPairs.join(edges).map(x => (x._2._2, x._2._1))).distinct().cache() nextCount = soPairs.count() } while (nextCount != oldCount) //soPairs = soPairs.sortByKey() def addKey2Val(key :Int, values: Iterable[Int]): List[Int] = { var res = ListBuffer[Int]() var i = values.iterator res += key while (i.hasNext) { res += i.next return res.toList var rootedRSG = var rootedRSG2 = soPairs rootedRSG.count var end = java.lang.System.currentTimeMillis(); print("Duration to rooted RSG = "+ (end-start)+" ms")quotesdbs_dbs45.pdfusesText_45

[PDF] MR - Spark - IGM

MR - Spark

Objectif Découvrir Apache Spark avec Scala

Installation

Manipulation

Exercice

1. text.count

2. Supprimer les lignes dont la longueur est nulle (length). On va utiliser la transformation Spark

3. A l'aide de la fonction Scala contains, on veut compter le nombre de 'CHAPTER' de l'ouvrage

4. On veut créer une nouvelle RDD qui va contenir le nombre de caractères par ligne. Donner la

5. On veut écrire un wordCount en Spark sur le texte. A partir du résultat de la question 2, on veut

5.a On veut créer une entrée dans une nouvelle RDD par mot, c'est-à-dire, qu'il faut

5b. A partir de la RDD obtenue en 5a, écrire le wordCount en vous inspirant de l'approche

6. On veut obtenir une liste ordonnée du résultat obtenu en question 5

6a. Ordonnée par mot (ordre alphabétique). Fournir le premier mot et le nombre

6b. Ordonnée par nombre décroissant d'occurrences. Fournir le mot apparaissant le plus

7. Nous n'avons pas nettoyé le texte avant le traitement, cela explique qu'il y a plusieurs entrées

7a. Vous devez créer une RDD ne contenant que les entrées avec le terme ''Alice''.

Combien en trouvez vous.

7b. Faire le cumul des valeurs du RDD obtenue en 7a.

8. Nous allons changer de jeu de données pour étudier les possibilités de jointure avec Spark.

8a. Nous allons charger les jeux de données CIP_GENER_bdpm.txt et CIS_CPD_bdpm.txt

8b.Dans les 2 fichiers, les séparateurs de champs sont '\t'. Pour le fichier gen, on ne

8c. Nous voulons obtenir la liste des codes CIS et de l'intitulé de la liste pour les

RDDs. Donner le nombre de lignes du résultat.

8d. Avons-nous des doublons de codes CIS dans la RDD résultante de 8c ?

8e. Combien y-a-t-il de doublons ?

9. On considère le RDD suivant :

9a. Créer un RDD ne contenant que les sujets et objets du graphe.

9b. Trouver les noeuds du graphe qui sont des racines (pas d'arc entrant sur ces noeuds).

9c. Créer un RDD qui contiendra l'ensemble des noeuds accessibles depuis chaque noeud

9d. A partir de ce dernier RDD et du RDD contenant les racines, former un RDD (rootedRSG) qui

Solution

1. text.count

2. val text2 = text.filter(x=>x.length()>0)

3. val chap = text.filter(x=>x.contains("CHAPTER"))

4. val l = text2.map(x=>x.length()).reduce((a,b)=>a.max(b))

5a. val words = text2.flatMap(x=>x.split(" "))

5b. val wordCount = words.map(x=>(x,1)).reduceByKey(_+₎ wordCount.count // 5299

6a. val wcSorted = wordCount.sortByKey()

6b. val wcs2 = wordCount.map({case(w,n)=>(n,w)}).sortByKey()

7a. val alice = wcs2.filter({case(n,w)=>w.contains("Alice")})

7b. val aliceCount = alice.map({case(n,w)=>n})

8a. val cpd = sc.textFile("/home/oliv/Documents/CIS_CPD_bdpm.txt")

8b. val gen2 = gen.map(x=>x.split("\t")).map(t=>(t(2),t(3)))

8c. val cisList = gen2.filter({case(c,g)=>g.toLong==0}).join(cpd2)

8d. val cis = cisList.map({case(c,(t,l))=> c}).distinct.count

8e. val doubl = cis.map(x=>(x,1)).reduceByKey(_+_).filter{case(c,n)=>n>1}