Bases de Données / Databases

Site Web de l'équipe BD du LIP6 / LIP6 DB Web Site

Outils pour utilisateurs

Outils du site

Piste : • DIU • TME DTD • Bernd Amann
site:recherche:logiciels:rdfdist

Différences

Ci-dessous, les différences entre deux révisions de la page.

Lien vers cette vue comparative

Les deux révisions précédentesRévision précédente
Prochaine révision		Révision précédente
site:recherche:logiciels:rdfdist [11/05/2015 12:29] – [RDFdist] aminesite:recherche:logiciels:rdfdist [26/04/2017 13:06] (Version actuelle) – amann
Ligne 1: Ligne 1:
-===== RDFdist =====+{{indexmenu_n>1}} 
 + 
 +===== RDFdist :  RDF distribution approaches using Spark =====
  
 This wiki page provides information about the experiments RDF distribution approaches using [[http://spark.apache.org/|Spark]]. This wiki page provides information about the experiments RDF distribution approaches using [[http://spark.apache.org/|Spark]].
-This information consists of i) the source code for both data preparation and query evaluation, ii) the query workload and iii) the  description of datasets used in the experiments. +This information consists of i) the query workload ii) the source code for both data preparation and query evaluation, and iii) the  description of two datasets used in the experiments. 
-For the sake of reproducibility, each source code is provied as a script which can be directly executed in the spark shell. +For the sake of reproducibility, each source code is provided as a script which can be directly executed in the spark shell.
-=====Data preparation and Query evaluation Source Code=====+
  
-====Hash-based approaches====+=====Query workload==== 
 +We picked three queries from the [[http://swat.cse.lehigh.edu/projects/lubm/|LUBM ]] dataset, namely #2, #9 and #12 
 +which are referred to as Query 2, Query 3 and Query 4, respectively. 
 +We have created an addirional one referred to as Query 1. 
 +We also created two queries for the [[https://www.wikidata.org/wiki/Wikidata:Main_Page|Wikidata]] dataset which are referred to as Query 5 and Query 6.
  
 +===Query 1 (synthetic, LUBM)===
 +<code>
 +SELECT ?x ?y ?z 
 +WHERE
 +{?x lubm:advisor ?y.
 + ?y lubm:worksFor ?z.
 + ?z lubm:subOrganisation ?t
 +}
 +</code>
  
-<code>The following code presents the random hashing approach for the LUBM datasets, i.e., queries Q1, Q2, Q3 and Q4 are executed and evaluated. 
  
 +
 +
 +===Query 2 (#2 of LUBM)===
 +<code>
 +PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> 
 +PREFIX ub: <http://www.lehigh.edu/~zhp2/2004/0401/univ-bench.owl#> 
 +SELECT ?X, ?Y, ?Z 
 +WHERE 
 +{?X rdf:type ub:GraduateStudent . 
 +  ?Y rdf:type ub:University . 
 +  ?Z rdf:type ub:Department . 
 +  ?X ub:memberOf ?Z . 
 +  ?Z ub:subOrganizationOf ?Y . 
 +  ?X ub:undergraduateDegreeFrom ?Y}
 +</code>
 +
 +===Query 3 (#9 of LUBM)===
 +<code>
 +PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> 
 +PREFIX ub: <http://www.lehigh.edu/~zhp2/2004/0401/univ-bench.owl#> 
 +SELECT ?X, ?Y, ?Z 
 +WHERE 
 +{?X rdf:type ub:Student . 
 +  ?Y rdf:type ub:Faculty . 
 +  ?Z rdf:type ub:Course . 
 +  ?X ub:advisor ?Y . 
 +  ?Y ub:teacherOf ?Z . 
 +  ?X ub:takesCourse ?Z}
 +</code>
 +
 +===Query 4 (#12 of LUBM)===
 +<code>
 +PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> 
 +PREFIX ub: <http://www.lehigh.edu/~zhp2/2004/0401/univ-bench.owl#> 
 +SELECT ?X, ?Y 
 +WHERE 
 +{?X rdf:type ub:Chair . 
 +  ?Y rdf:type ub:Department . 
 +  ?X ub:worksFor ?Y . 
 +  ?Y ub:subOrganizationOf <http://www.University0.edu>} 
 +
 +</code>
 +
 +===Query 5 (synthetic, Wikidata)===
 +<code>
 +SELECT ?x ?y ?z 
 +WHERE  
 +{?x entity:P131s ?y. 
 +?y entity:P961v> ?z. 
 +?z entity:P704s ?w
 +}
 +</code>
 +
 +===Query 6 (synthetic, Wikidata) ===
 +<code>
 +SELECT ?x ?y ?z 
 +{?x entity:P39v ?y. 
 +?x entity:P580q ?z. 
 +?x rdf:type ?w
 +}
 +</code>
 +
 +=====Source Code=====
 +
 +
 +
 +====Hash-based approaches====
 +The following code corresponds to the triple hashing and subject hashing approaches for the LUBM datasets.
 +It consists of a data preparation part followd by a query evaluation part.
  
  
Ligne 28: Ligne 110:
  
 val t1 = java.lang.System.currentTimeMillis(); val t1 = java.lang.System.currentTimeMillis();
 +
 +
 +/**
 +* set inputData with the path to the data encoded as quadruples (see Datasets excerpts)
 +*/
  
 // loading and transformating the dataset // loading and transformating the dataset
Ligne 36: Ligne 123:
 println("Loading time "+ (t2 - t1) +" msec for "+part+" partitions"); println("Loading time "+ (t2 - t1) +" msec for "+part+" partitions");
  
-// Partitioning the dataset +// Partitioning the dataset  
-val triples = triples0.map{case(f,(s,p,o))=>((s+p+o),(s,p,o))}.partitionBy(new HashPartitioner(part))+/** 
 +* Uncomment one of the following lines depending on whether hashing is applied 
 +on the entire triple or only the subject  
 +*/ 
 +// val triples = triples0.partitionBy(new HashPartitioner(part))  
 +// val triples = triples0.map{case(f,(s,p,o))=>((s+p+o),(s,p,o))}.partitionBy(new HashPartitioner(part)) 
  
 //triples.persist //triples.persist
Ligne 63: Ligne 156:
 val takco : Long = 1115684864 val takco : Long = 1115684864
  
-def ajout(a : ListBuffer[(Long, Long, Long)], e: (Long, Long, Long) ) : ListBuffer[(Long, Long, Long)] = { 
-  a += e 
-  return a 
-} 
  
     // -----------------------------------------------------------     // -----------------------------------------------------------
Ligne 144: Ligne 233:
  
 println("Processing LUBM #12 "+ (t2 - t1) +" msec for "+part+" partitions"); println("Processing LUBM #12 "+ (t2 - t1) +" msec for "+part+" partitions");
 +
 +// -----------------------------------------------------------
 +// highly selective on wikidata :: 8 results 
 +    // QR5:  ?x <http://www.wikidata.org/entity/P131s> ?y. 
 +    ?y <http://www.wikidata.org/entity/P961v> ?z. 
 +    ?z <http://www.wikidata.org/entity/P704s> ?w. 
 +    // -----------------------------------------------------------
 +val qr5 = triples.filter({case(s,p,o)=>p==P131s}).map({case(s,p,o)=>(o,(s,p))}).
 +    join(triples.filter({case(s,p,o)=>p==P961v}).map({case(s,p,o)=>(s,(p,o))})).
 +    map({case(k,((s1,p1),(p2,o2)))=>(o2,s1)}).
 +    join(triples.filter({case(s,p,o)=>p==P704v}).map({case(s,p,o)=>(s,(p,o))}))
 +qr5.persist
 +qr5.collect 
 +
 +// mid selectivity start shaped query : 10418 results
 +// QR6: ?x <http://www.wikidata.org/entity/P39v> ?y. ?x <http://www.wikidata.org/entity/P580q> ?z. ?x <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> ?w
 +
 +val qr6 = triples.filter({case(s,p,o)=>p==P39v}).map({case(s,p,o)=>(s,null)}).
 +    join(triples.filter({case(s,p,o)=>p==P580q}).map({case(s,p,o)=>(s,null)})).
 +    join(triples.filter({case(s,p,o)=>p==rdftype}))
 +
 +qr6.persist
 +qr6.collect 
 +
 </code> </code>
  
-==Hashing applied on the subject == 
  
-====Partitioning-based approaches==== 
  
-==nHop==+====Graph partitioning-based approaches====
  
-==WARP==+===Huang Approach === 
 +<code> 
 +import org.apache.spark.HashPartitioner 
 +import scala.collection.mutable.ListBuffer
  
-==Hybrid==+val folder=  "lubm" //"watdiv"  
 +val dataset= "univ" //"watdiv"  
 +val scale="1k" 
 +val part=20 //10, 20
  
 +val folderName = folder +scale
 +val fileName = dataset+scale+"_encoded_unique_quads.part."+part+".2hop"
 +
 +val t1 = java.lang.System.currentTimeMillis();
 +
 +val quads_I_SPO = sc.textFile(s"/user/olivier/${folderName}/${fileName}").coalesce(part).map(x=>x.split(",")).map(t=>(t(3).replace(")","").toLong, (t(0).replace("(","").toLong,t(1).toLong,t(2).toLong)))
 +
 +val quadsDist = quads_I_SPO.partitionBy(new HashPartitioner(part)).persist
 +
 +
 +
 +val t2 = java.lang.System.currentTimeMillis();
 +
 +print("Loading time of quads : "+(t2-t1)/1000 +" sec")
 +
 +    val advisor : Long = 1233125376
 +    val worksFor : Long = 1136656384
 +    val suborg : Long = 1224736768
 +    val memof : Long = 113246208
 +    val undeg : Long = 1101004800
 +    val teaof : Long = 1199570944
 +    val takco : Long = 1115684864
 +
 +
 +    // -----------------------------------------------------------
 +    // Query 1 : (not part of the benchmark)
 +    // Pattern: (x advisor y) (y worksFor z) (z subOrganisation t)
 +    // -----------------------------------------------------------
 +
 +var t1 = java.lang.System.currentTimeMillis();
 +
 +var pataws = quadsDist.filter({case(i,(s,p,o)) => p==advisor}).map({case(i,(s,p,o)) => (o,s)}).
 +    join(quadsDist.filter({case(i,(s,p,o)) => p==worksFor}).map({case(i,(s,p,o)) => (s,o)}),part).
 +    map({case (y,(x,z)) => (z,(x,y))}).
 +    join(quadsDist.filter({case(i,(s,p,o)) => p==suborg}).map({case(i,(s,p,o)) => (s,o)}), part).
 +    map({case (z,((x,y),t)) => (x,y,z,t)})
 +
 +pataws.count
 +
 +var pataws2 = pataws.flatMap(x=>x).distinct
 +
 +var t2= java.lang.System.currentTimeMillis();
 +
 +println("Processing Q1 "+ (t2 - t1) +" msec for "+part+" partitions");
 +
 +
 +    // -----------------------------------------------------------
 +    // LUBM 2 : MSU
 +    // Pattern: (x memberOf y) (y subOrg z) (x UndergraduateDegreeFrom z)
 +    // -----------------------------------------------------------
 +
 +var t1 = java.lang.System.currentTimeMillis();
 +
 +//var pmemof = quadsDist.filter({case(i,(s,p,o)) => p==memof}).cache()
 +
 +var patmsu = quadsDist.filter({case(i,(s,p,o)) => p==memof}).map({case(i,(s,p,o)) => (o,s)}).
 +             join(quadsDist.filter({case(i,(s,p,o)) => p==suborg}).map({case(i,(s,p,o)) => (s,o)}),part).
 +             map({case (y,(x,z)) => (x+""+z,(x,y,z))}). 
 +             join(quadsDist.filter({case(i,(s,p,o)) => p==undeg}).map({case(i,(x,p,z))=> (x+""+z,null)}))
 +
 +patmsu.count
 +
 +var patmsu2 = patmsu.flatMap(identity).distinct
 +
 +var t2= java.lang.System.currentTimeMillis();
 +
 +println("Processing Q2 "+ (t2 - t1) +" msec for "+part+" partitions");
 +
 +    // -----------------------------------------------------------
 +    // LUBM 9 : ATT
 +    // Pattern: (x advisor y) (y teacherOf z) (x takesCourse z)
 +    // -----------------------------------------------------------
 +
 +var t1 = java.lang.System.currentTimeMillis();
 +
 +var patatt = quadsDist.filter({case(i,(s,p,o)) => p==advisor}).map({case(i,(s,p,o)) => (o,s)}).
 +              join(quadsDist.filter({case(i,(s,p,o)) => p==teaof}).map({case(i,(s,p,o)) => (s,o)}), part).
 +              map({case (y,(x,z)) => (x+""+z,(x,y,z))}).
 +              join(quadsDist.filter({case(i,(s,p,o)) => p==takco}).map({case(i,(s,p,o))=> (s+""+o,null)}), part)
 +
 +patatt.distinct.count
 +
 +var t2= java.lang.System.currentTimeMillis();
 +
 +println("Processing Q3 (LUBM #9) "+ (t2 - t1) +" msec for "+part+" partitions");
 +
 +
 +    // -----------------------------------------------------------
 +    // LUBM 12 : WS
 +    // Pattern: (y worksFor z) (z subOrganisation t)
 +    // -----------------------------------------------------------
 +
 +val t1 = java.lang.System.currentTimeMillis();
 +
 +val patws = quadsDist.filter({case(i,(s,p,o)) => p==worksFor}).map({case(i,(s,p,o)) => ((i,o),s)}).join(quadsDist.filter({case(i,(s,p,o)) => p==suborg}).map({case(i,(s,p,o)) => ((i,s),o)}),part).map({case ((i,k),(s,o)) => (s,o)})
 +
 +val ans_patws = patws.distinct.count
 +
 +val t2= java.lang.System.currentTimeMillis();
 +
 +println("Processing LUBM #12 "+ (t2 - t1) +" msec for "+part+" partitions");
 +</code>
 +
 +===Warp===
 +<code>
 +// Spark implementation of WARP replication
 +// usage: run this code into the spark-shell
 +
 +import scala.collection.mutable.ListBuffer
 +import org.apache.spark.rdd.RDD
 +import scala.reflect.ClassTag
 +import org.apache.spark.SparkContext
 +import org.apache.spark.SparkContext._
 +
 +import org.apache.spark.HashPartitioner
 +import org.apache.spark.Partitioner
 +
 +import java.io.Serializable
 +
 +
 +val folder= "lubm" //"watdiv" 
 +val dataset= "univ" //"watdiv"//
 +val scale="1k"
 +
 +/* We have 15 cores per machine. 
 +   Each core is accessing a separate part in parallel.
 +   The default parallelism = #machines * 15 cores
 +   <=>   #machines = defaultParallelism / 15
 +*/
 +val machine = sc.defaultParallelism /15
 +val part =  sc.defaultParallelism  
 +val folderName = folder + scale
 +val fileName = dataset + scale
 +/**
 +* set inputData with the path to the data encoded as quadruples (see Datasets excerpts)
 +*/
 +val inputData = s"/user/olivier/${folderName}/${fileName}_encoded_unique_quads.part.${machine}"
 +
 +// Initial state, delete the storage
 +sc.getPersistentRDDs.values.foreach(x => x.unpersist())
 +sc.getPersistentRDDs.values.size
 +
 +var t1 = java.lang.System.currentTimeMillis();
 +
 +def affiche(s: String): Unit={
 +println("#### ------------------------------- ####")
 +println(s)
 +println("#### ------------------------------- ####")
 +}
 +
 +
 +/*
 +-----------------------
 +STEP 1: read triples
 +-----------------------
 +*/
 +
 +/**
 +  Function  lireTriples reads the input data
 +  returns a quad (subject, property, object, partID)
 +  with partID = hash(subject)
 +**/
 +def lireTriples(input: String): RDD[(Long, Long, Long, Int)] = {
 +   return sc.textFile(input).
 +    map(line => line.substring(1, line.length -1).split(",")).
 +    map(tab => (tab(0).toLong, tab(1).toLong, tab(2).toLong, (tab(0).toInt).hashCode%machine))
 +}
 +
 +val triples = lireTriples(inputData)
 +triples.setName("triples")
 +triples.persist
 +
 +//nombre total de nuplets
 +//triples.count
 +
 +// stat: nb triplets par partition
 +//triples.map{ case(s,p,o,f) => (f,1)}.reduceByKey( (a,b) => a+b).collect.foreach(println)
 +
 +
 +
 +/*
 +---------------------------------------------------------------------------
 +STEP 2: COMPUTE REPLICATION TRIPLES
 +
 +Computes the triple to replicate in every partition, for each query
 +---------------------------------------------------------------------------
 +*/
 +
 +/*
 + Fonction filterProp : selects the triples a a given property
 + */
 +def filterProp(input: RDD[(Long, Long, Long, Int)], prop: Long): RDD[(Long, Long, Int)] = {
 +x  // on projette sur S et O car P est fixe
 +  return input.filter(x => x._2 == prop).map{ case (s,p,o,f) => (s,o,f)}
 +}
 +
 +def getOutsideTriple(seedF: Int, t: ListBuffer[(Long, Long, Int)], properties: Array[Long]): Seq[(Long, Long, Long, Int)] = {
 +  var a = t.toArray
 +  var res: ListBuffer[(Long, Long, Long, Int)] = ListBuffer()
 +
 +  for(i <-0 to (a.length - 1)) {
 +    if( a(i)._3 != seedF) {
 +      res.append( (a(i)._1,  properties(i), a(i)._2,  seedF))
 +    }
 +  }
 +  return res
 +}
 +
 +def getReplicaForQuery(query: RDD[ListBuffer[(Long, Long, Int)]], properties: Array[Long], nbCandidateSeed: Int): RDD[(Long, Long, Long, Int)] = {
 +  var replica = sc.parallelize(List((0L,0L,0L,0)))
 +  var min_nb = -1L
 +
 +  for (i <- 0 to (nbCandidateSeed-1)) {
 +    var t1 = query.map(x => (x(i)._3, x))
 +    //t1.take(5).foreach(println)
 +
 +    // lister les triplets a  repliquer = ceux qui ne sont pas dans la seed partition
 +    var t2 = t1.flatMap{ case(seedF, tripleList) => getOutsideTriple(seedF, tripleList, properties)}.distinct
 +
 +    // count the triples to replicate
 +    var nb = t2.count
 +    if(min_nb == -1 || nb < min_nb){
 +      min_nb=nb
 +      replica=t2
 +      println("current:")
 +      println(s"seed $i, replica count: $nb ")
 +    }
 +    else {
 +      println(s"-----ignore seed $i, replica count: $nb ")
 +    }
 +  }
 +  return replica
 +}
 +
 +// -------------------------------------------------------
 +//  Compute replication for the folowing QUERIES 
 +// -------------------------------------------------------
 +
 +// based on dictionnary encoding
 +val advisor : Long = 1233125376
 +val worksFor : Long = 1136656384
 +val suborg : Long = 1224736768
 +val memof : Long = 1132462080
 +val undeg : Long = 1101004800
 +val teaof : Long = 1199570944
 +val takco : Long = 1115684864
 +
 +// -----------------------------------------------------------
 +// Query 1 : (not part of the benchmark)
 +// Pattern: (x advisor y) (y worksFor z) (z subOrganisation t)
 +// -----------------------------------------------------------
 +val query1 = filterProp(triples, advisor).map{ case (x, y, f1) => ( y, (x, f1))}.
 +  join(filterProp(triples, worksFor).map{ case (y, z, f2) => ( y, (z, f2) )}).
 +  map{ case ( y, ((x,f1), (z, f2))) => (z, (x, f1, y, f2)) }.
 +  join(filterProp(triples, suborg).map{ case ( z, t, f3) => (z, (t,f3))}).
 +  map{ case ( z, ( (x,f1,y,f2), (t,f3) )) => ListBuffer( (x, y, f1), (y, z, f2) , (z, t, f3))}
 +
 +query1.setName("q1")
 +//query1.persist()
 +//query1.count
 +// resultat: 4108791
 +
 +var t1 = java.lang.System.currentTimeMillis();
 +
 +val replica1 =  getReplicaForQuery(query1, Array(advisor, worksFor, suborg), 3)
 +
 +var t2= java.lang.System.currentTimeMillis();
 +
 +println("Processing Q1 "+ (t2 - t1) +" msec for "+part+" partitions");
 +
 +// stat: nb triples a repliquer dans chaque partition
 +//replica1.map{ case(s,p,o,f) => (f,1)}.reduceByKey( (a,b) => a+b).collect.foreach(println)
 +// (0,15551) (1,15776) (2,17639) (3,201275) (4,46068)
 +
 +
 +// -----------------------------------------------------------
 +// LUBM 2 : MSU
 +// Pattern: (x memberOf y) (y subOrg z) (x UndergraduateDegreeFrom z)
 +// -----------------------------------------------------------
 +val query2 = filterProp(triples, memof).map{ case (x, y, f1) => ( y, (x, f1))}.
 +  join(filterProp(triples, suborg).map{ case (y, z, f2) => ( y, (z, f2) )}).
 +  map{ case ( y, ((x,f1), (z,f2) )) => ((x, z), (y, f1, f2)) }.
 +  join(filterProp(triples, undeg).map{ case (x, z, f3) => ( (x, z), f3)}).
 +  map{ case ((x, z), ((y, f1, f2), f3)) => ListBuffer((x, y, f1), (y, z, f2), (x, z, f3)) }
 +
 +query2.setName("q2")
 +//query2.persist()
 +//query2.count
 +
 +val replica2 =  getReplicaForQuery(query2, Array(memof, suborg, undeg), 2)
 +
 +//stat: nb triples a  repliquer dans chaque partition
 +//replica2.map{ case(s,p,o,f) => (f,1)}.reduceByKey( (a,b) => a+b).collect.foreach(println)
 +//(0,1) (1,3) (3,4) (4,896)
 +
 +// -----------------------------------------------------------
 +// LUBM 9 : ATT
 +// Pattern: (x advisor y) (y teacherOf z) (x takesCourse z)
 +// -----------------------------------------------------------
 +val query9 = filterProp(triples, advisor).map{ case (x, y, f1) => ( y,(x, f1))}.
 +  join(filterProp(triples, teaof).map{ case (y, z, f2) => ( y, (z, f2) )}).
 +  map{ case ( y, ((x, f1), (z, f2) )) => ((x, z), (y, f1, f2)) }.
 +  join(filterProp(triples, takco).map{ case (x, z, f3) => ( (x, z), f3)}).
 +  map{ case ((x, z), ((y, f1, f2), f3)) => ListBuffer((x, y, f1), (y, z, f2), (x, z, f3)) }
 +
 +query9.setName("q9")
 +//query9.persist()
 +query9.count
 +// 272982
 +
 +val replica9 =  getReplicaForQuery(query9, Array(advisor, teaof, takco) ,2)
 +
 +// stat : nb triples à repliquer dans chaque partition
 +//replica9.map{ case(s,p,o,f) => (f,1)}.reduceByKey( (a,b) => a+b).collect.foreach(println)
 +//(0,3131) (1,3213) (2,3504) (3,44706) (4,2014)
 +
 +
 +// -----------------------------------------------------------
 +// LUBM 12 : WS
 +// Pattern: (y worksFor z) (z subOrganisation t)
 +// -----------------------------------------------------------
 +val query12 = filterProp(triples, worksFor).map{ case (y, z, f1) => ( z, (y, f1))}.
 +  join(filterProp(triples, suborg).map{ case (z, t, f2) => ( z, (t, f2))}).
 +  map{ case ( z, ((y, f1), (t, f2) )) => ListBuffer((y, z, f1), (z, t, f2)) }
 +
 +query12.setName("q12")
 +//query12.persist()
 +query12.count
 +//720628
 +
 +val replica12 =  getReplicaForQuery(query12, Array(worksFor, suborg), 2)
 +
 +//stat : nb triples à repliquer dans chaque partition
 +//replica12.map{ case(s,p,o,f) => (f,1)}.reduceByKey( (a,b) => a+b).collect.foreach(println)
 +
 +
 +/*
 +---------------------------------------------------------------------------
 +STEP 3: UNION DES REPLICAS CALCULES POUR CHAQUE REQUETE 
 +---------------------------------------------------------------------------
 +*/
 +
 +val allreplica = replica1.union(replica2).union(replica9).union(replica12).distinct
 +
 +val nbAjout = allreplica.count
 +// 357 161 pour 5 part
 +// 734 295 pour 10 part
 +// 779 983 pour 20 part
 +
 +// replication rate
 +affiche("N ajout:  " + nbAjout.toDouble)
 +affiche("taux de replication: " + (nbAjout.toDouble / triples.count))
 +
 +var t2= java.lang.System.currentTimeMillis();
 +affiche("Preparation time "+ (t2 - t1).toDouble/1000 +" sec for " + part + " partitions");
 +
 +val oDist = triples.map{ case(s,p,o,f) => (f,1)}.reduceByKey( (a,b) => a+b)
 +val oMean = oDist.map{case(f,c)=>c}.sum / machine
 +val odevs = oDist.map{case(f,c)=>c}.map(score => (score - oMean) * (score - oMean))
 +val ostddev = Math.sqrt(odevs.sum / machine)
 +
 +val nDist = oDist.join(allreplica.map{ case(s,p,o,f) => (f,1)}.reduceByKey( (a,b) => a+b)).map{case(f,(oc,on))=>(oc+on)}
 +val nMean = nDist.sum / machine
 +val ndevs = nDist.map(score => (score - nMean) * (score - nMean))
 +val nstddev = Math.sqrt(ndevs.sum / machine)
 + 
 +
 +
 +/*
 +---------------------------------------------------------------------------
 +STEP 4: LOCAL QUERY PROCESSING 
 +On the WARP replicated data
 +---------------------------------------------------------------------------
 +*/
 +
 +// Extends each partition with the WARP replicated triples.
 +// Each machine stores one partition
 +val triplesWARP = triples.union(allreplica).
 +  map{ case (s,p,o,f) => (f, (s,p,o))}.
 +  partitionBy(new HashPartitioner(machine)).map{ case (f, (s,p,o)) => (s,p,o,f)}
 +
 +triplesWARP.setName("triples WARP")
 +triplesWARP.persist()
 +triplesWARP.count
 +
 +
 +var t2= java.lang.System.currentTimeMillis();
 +affiche("Preparation time "+ (t2 - t1).toDouble/1000 +" sec for " + part + " partitions");
 +
 +
 +// -------------------
 +//  Q1 LOCAL
 +// -------------------
 +t1= java.lang.System.currentTimeMillis();
 +
 +// Query1 locale avec projection sur les variables (x,y,z,t)
 +val localQuery1 = filterProp(triplesWARP, advisor).map{ case (x, y, f) => ( (y, f), x)}.
 +  join(filterProp(triplesWARP, worksFor).map{ case (y, z, f) => ( (y, f), z )}).
 +  map{ case ( (y, f), (x, z)) => ((z, f), (x, y)) }.
 +  join(filterProp(triplesWARP, suborg).map{ case ( z, t, f) => ((z, f), t)}).
 +  map{ case ( (z, f), ( (x,y), t)) => (x, y, z, t)}
 +
 +localQuery1.count
 +//4108791  OK idem sans replication
 +
 +t2= java.lang.System.currentTimeMillis();
 +affiche("Duréee de la requête Q1 : "+ (t2 - t1) +" msec pour " + part + " partitions");
 +
 +// -----------------------------------------------------------
 +// Q2 LOCAL:
 +// Pattern: (x memberOf y) (y subOrg z) (x UndergraduateDegreeFrom z)
 +// -----------------------------------------------------------
 +t1= java.lang.System.currentTimeMillis();
 +
 +val localQuery2 = filterProp(triplesWARP, memof).map{ case (x, y, f) => ( (y, f), x)}.
 +  join(filterProp(triplesWARP, suborg).map{ case (y, z, f) => ( (y, f), z )}).
 +  map{ case ( (y, f), (x, z)) => ((x, z, f), y) }.
 +  join(filterProp(triplesWARP, undeg).map{ case (x, z, f) => ( (x, z, f), 1)}).
 +  map{ case ((x, z, f), (y, 1)) => (x, y, z) }
 +
 +localQuery2.count
 +//2528
 +t2= java.lang.System.currentTimeMillis();
 +affiche("Duree de la requete Q2 : "+ (t2 - t1) +" msec pour " + part + " partitions");
 +
 +
 +// -----------------------------------------------------------
 +// Q9 LOCAL
 +// Pattern: (x advisor y) (y teacherOf z) (x takesCourse z)
 +// -----------------------------------------------------------
 +t1= java.lang.System.currentTimeMillis();
 +
 +val localQuery9 = filterProp(triplesWARP, advisor).map{ case (x, y, f) => ( (y, f), x)}.
 +  join(filterProp(triplesWARP, teaof).map{ case (y, z, f) => ( (y, f), z )}).
 +  map{ case ( (y, f), (x, z)) => ((x, z, f), y) }.
 +  join(filterProp(triplesWARP, takco).map{ case (x, z, f) => ( (x, z, f), 1)}).
 +  map{ case ((x, z, f), (y, 1)) => (x, y, z) }
 +
 +localQuery9.count
 +//272982
 +
 +t2= java.lang.System.currentTimeMillis();
 +affiche("Duree de la requete Q9 : "+ (t2 - t1) +" msec pour " + part + " partitions");
 +
 +
 +// -----------------------------------------------------------
 +// Q12 LOCAL
 +// Pattern: (y worksFor z) (z subOrganisation t)
 +// -----------------------------------------------------------
 +t1= java.lang.System.currentTimeMillis();
 +
 +val localQuery12 = filterProp(triplesWARP, worksFor).map{ case (y, z, f) => ( (z, f), y)}.
 +  join(filterProp(triplesWARP, suborg).map{ case (z, t, f) => ( (z, f), t)}).
 +  map{ case ( (z, f), (y, t)) => (y, z, t) }
 +
 +localQuery12.count
 +//938356
 +
 +t2= java.lang.System.currentTimeMillis();
 +affiche("Duree de la requete Q12: "+ (t2 - t1) +" msec pour " + part + " partitions");
 +
 +
 +</code>
 +===2-hop based approach===
 +<code>
 +val folder=  "lubm"
 +val dataset= "univ"
 +val scale="10k"
 +
 +val folderName = folder +scale
 +val part = Array(5,10,20)
 +
 +for (p <- part)
 +{
 +val fileName = dataset+scale+"_encoded_unique_quads.part."+p
 +val fileNamewatdiv2k_encoded_unique_quads.partNew.5
 +val t1 = java.lang.System.currentTimeMillis();
 +
 +val quads = sc.textFile(s"/user/olivier/${folderName}/${fileName}").map(x=>x.split(",")).map(t=>(t(0).replace("(","").toLong,t(1).toLong,t(2).toLong,t(3).replace(")","").toLong))
 +
 +var addOneHop = quads.map({case(s,p,o,i)=>(o,i)}).join(quads.map({case(s,p,o,i)=>(s,(p,o,i))})).filter({case(termS,(i1,(p,o,i2)))=>i1!=i2}).distinct.map({case(termS,(i1,(p,o,i2)))=>(termS,p,o,i1)})
 +
 +val newQuads = quads.union(addOneHop).distinct
 +val newQuadsSize = newQuads.count
 +
 +val t2 = java.lang.System.currentTimeMillis();
 +val hopSize = addOneHop.count
 +println(s"Time to compute one more hop on $folderName for $p partitions is ${t2-t1}")
 +println(s" new size = $newQuadsSize , added $hopSize")
 +newQuads.saveAsTextFile(s"/user/olivier/${folderName}/${fileName}.2hop")
 +} 
 +</code>
 ====Datasets excerpts==== ====Datasets excerpts====
 +===Encoding of LUBM concepts and properties===
 +
 +<code>
 +Properties:
 +0 <http://www.w3.org/1999/02/22-rdf-syntax-ns#type>
 +603979776 <http://www.univ-mlv.fr/~ocure/lubm.owl#officeNumber>
 +671088640 <http://www.univ-mlv.fr/~ocure/lubm.owl#name>
 +738197504 <http://www.univ-mlv.fr/~ocure/lubm.owl#title>
 +805306368 <http://www.univ-mlv.fr/~ocure/lubm.owl#age>
 +872415232 <http://www.univ-mlv.fr/~ocure/lubm.owl#telephone>
 +939524096 <http://www.univ-mlv.fr/~ocure/lubm.owl#emailAddress>
 +1006632960 <http://www.univ-mlv.fr/~ocure/lubm.owl#researchInterest>
 +1082130432 <http://www.univ-mlv.fr/~ocure/lubm.owl#researchProject>
 +1090519040 <http://www.univ-mlv.fr/~ocure/lubm.owl#hasAlumnus>
 +1098907648 <http://www.univ-mlv.fr/~ocure/lubm.owl#degreeFrom>
 +1101004800 <http://www.univ-mlv.fr/~ocure/lubm.owl#undergraduateDegreeFrom>
 +1103101952 <http://www.univ-mlv.fr/~ocure/lubm.owl#mastersDegreeFrom>
 +1105199104 <http://www.univ-mlv.fr/~ocure/lubm.owl#doctoralDegreeFrom>
 +1107296256 <http://www.univ-mlv.fr/~ocure/lubm.owl#orgPublication>
 +1115684864 <http://www.univ-mlv.fr/~ocure/lubm.owl#takesCourse>
 +1124073472 <http://www.univ-mlv.fr/~ocure/lubm.owl#member>
 +1132462080 <http://www.univ-mlv.fr/~ocure/lubm.owl#memberOf>
 +1136656384 <http://www.univ-mlv.fr/~ocure/lubm.owl#worksFor>
 +1138753536 <http://www.univ-mlv.fr/~ocure/lubm.owl#headOf>
 +1140850688 <http://www.univ-mlv.fr/~ocure/lubm.owl#teachingAssistantOf>
 +1149239296 <http://www.univ-mlv.fr/~ocure/lubm.owl#listedCourse>
 +1157627904 <http://www.univ-mlv.fr/~ocure/lubm.owl#softwareDocumentation>
 +1166016512 <http://www.univ-mlv.fr/~ocure/lubm.owl#publicationAuthor>
 +1174405120 <http://www.univ-mlv.fr/~ocure/lubm.owl#softwareVersion>
 +1182793728 <http://www.univ-mlv.fr/~ocure/lubm.owl#affiliateOf>
 +1191182336 <http://www.univ-mlv.fr/~ocure/lubm.owl#tenured>
 +1199570944 <http://www.univ-mlv.fr/~ocure/lubm.owl#teacherOf>
 +1207959552 <http://www.univ-mlv.fr/~ocure/lubm.owl#publicationDate>
 +1216348160 <http://www.univ-mlv.fr/~ocure/lubm.owl#affiliatedOrganizationOf>
 +1224736768 <http://www.univ-mlv.fr/~ocure/lubm.owl#subOrganizationOf>
 +1233125376 <http://www.univ-mlv.fr/~ocure/lubm.owl#advisor>
 +1241513984 <http://www.univ-mlv.fr/~ocure/lubm.owl#publicationResearch>
 +
 +Concepts:
 +0 <http://www.univ-mlv.fr/~ocure/lubm.owl#Schedule>
 +268435456 <http://www.univ-mlv.fr/~ocure/lubm.owl#Organization>
 +301989888 <http://www.univ-mlv.fr/~ocure/lubm.owl#College>
 +335544320 <http://www.univ-mlv.fr/~ocure/lubm.owl#Department>
 +369098752 <http://www.univ-mlv.fr/~ocure/lubm.owl#Institute>
 +402653184 <http://www.univ-mlv.fr/~ocure/lubm.owl#ResearchGroup>
 +436207616 <http://www.univ-mlv.fr/~ocure/lubm.owl#Program>
 +469762048 <http://www.univ-mlv.fr/~ocure/lubm.owl#University>
 +536870912 <http://www.univ-mlv.fr/~ocure/lubm.owl#Publication>
 +570425344 <http://www.univ-mlv.fr/~ocure/lubm.owl#Software>
 +603979776 <http://www.univ-mlv.fr/~ocure/lubm.owl#Book>
 +637534208 <http://www.univ-mlv.fr/~ocure/lubm.owl#Specification>
 +671088640 <http://www.univ-mlv.fr/~ocure/lubm.owl#Manual>
 +704643072 <http://www.univ-mlv.fr/~ocure/lubm.owl#Article>
 +713031680 <http://www.univ-mlv.fr/~ocure/lubm.owl#TechnicalReport>
 +721420288 <http://www.univ-mlv.fr/~ocure/lubm.owl#ConferencePaper>
 +729808896 <http://www.univ-mlv.fr/~ocure/lubm.owl#JournalArticle>
 +738197504 <http://www.univ-mlv.fr/~ocure/lubm.owl#UnofficialPublication>
 +805306368 <http://www.univ-mlv.fr/~ocure/lubm.owl#Person>
 +872415232 <http://www.univ-mlv.fr/~ocure/lubm.owl#TeachingAssistant>
 +939524096 <http://www.univ-mlv.fr/~ocure/lubm.owl#Student>
 +956301312 <http://www.univ-mlv.fr/~ocure/lubm.owl#GraduateStudent>
 +973078528 <http://www.univ-mlv.fr/~ocure/lubm.owl#UndergraduateStudent>
 +1006632960 <http://www.univ-mlv.fr/~ocure/lubm.owl#Employee>
 +1015021568 <http://www.univ-mlv.fr/~ocure/lubm.owl#ResearchAssistant>
 +1023410176 <http://www.univ-mlv.fr/~ocure/lubm.owl#Director>
 +1031798784 <http://www.univ-mlv.fr/~ocure/lubm.owl#AdministrativeStaff>
 +1033895936 <http://www.univ-mlv.fr/~ocure/lubm.owl#SystemsStaff>
 +1035993088 <http://www.univ-mlv.fr/~ocure/lubm.owl#ClericalStaff>
 +1040187392 <http://www.univ-mlv.fr/~ocure/lubm.owl#Faculty>
 +1042284544 <http://www.univ-mlv.fr/~ocure/lubm.owl#PostDoc>
 +1044381696 <http://www.univ-mlv.fr/~ocure/lubm.owl#Professor>
 +1044643840 <http://www.univ-mlv.fr/~ocure/lubm.owl#Chair>
 +1044905984 <http://www.univ-mlv.fr/~ocure/lubm.owl#VisitingProfessor>
 +1045168128 <http://www.univ-mlv.fr/~ocure/lubm.owl#AssociateProfessor>
 +1045430272 <http://www.univ-mlv.fr/~ocure/lubm.owl#Dean>
 +1045692416 <http://www.univ-mlv.fr/~ocure/lubm.owl#FullProfessor>
 +1045954560 <http://www.univ-mlv.fr/~ocure/lubm.owl#AssistantProfessor>
 +1046478848 <http://www.univ-mlv.fr/~ocure/lubm.owl#Lecturer>
 +1073741824 <http://www.univ-mlv.fr/~ocure/lubm.owl#Work>
 +1140850688 <http://www.univ-mlv.fr/~ocure/lubm.owl#Course>
 +1174405120 <http://www.univ-mlv.fr/~ocure/lubm.owl#GraduateCourse>
 +1207959552 <http://www.univ-mlv.fr/~ocure/lubm.owl#Research> 
 +</code>
 +
 +===LUBM Univ1  ===
 +-[[http://webia.lip6.fr/~baazizi/research/iswc2015eval/sources/univ1_encoded_unique.id|encoded triples]](2.1MB)
 +
 +-[[http://webia.lip6.fr/~baazizi/research/iswc2015eval/sources/quads_plus_replicas.id|encoded quaruples with replication]](2.3MB)
 +
 +-[[http://webia.lip6.fr/~baazizi/research/iswc2015eval/sources/quads.id|replicated quaruples]](0.3MB)
 +
 +
  
 +-[[http://webia.lip6.fr/~baazizi/research/iswc2015eval/sources/univ1.nt|univ1.nt]] (16.3MB)
site/recherche/logiciels/rdfdist.1431340141.txt.gz · Dernière modification : de amine