This wiki page provides information about the experiments RDF distribution approaches using Spark. 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 provided as a script which can be directly executed in the spark shell.
We picked three queries from the 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 Wikidata dataset which are referred to as Query 5 and Query 6.
SELECT ?x ?y ?z
WHERE
{?x lubm:advisor ?y.
?y lubm:worksFor ?z.
?z lubm:subOrganisation ?t
}
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}
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}
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>}
SELECT ?x ?y ?z
WHERE
{?x entity:P131s ?y.
?y entity:P961v> ?z.
?z entity:P704s ?w
}
SELECT ?x ?y ?z
{?x entity:P39v ?y.
?x entity:P580q ?z.
?x rdf:type ?w
}
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.
import org.apache.spark.HashPartitioner
import scala.collection.mutable.ListBuffer
val folder= "lubm"
val dataset= "univ"
val scale="10k"
val coreNumber = 20
val machine = sc.defaultParallelism /coreNumber
val part = sc.defaultParallelism
val folderName = folder + scale
val fileName = dataset + scale
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
val triples0 = sc.textFile(s"/user/olivier/${folderName}/${fileName}_encoded_unique").map(x => x.split(" ")).map(t => ((t(0).toLong+t(1).toLong+t(2).toLong).hashCode%machine,(t(0).toLong,t(1).toLong, t(2).toLong))).persist
triples0.count
val t2= java.lang.System.currentTimeMillis();
println("Loading time "+ (t2 - t1) +" msec for "+part+" partitions");
// Partitioning the dataset
/**
* 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.count
val t3= java.lang.System.currentTimeMillis();
println("Partitioning time "+ (t3 - t2) +" msec for "+part+" partitions");
val oDist = triples.map{ case(f,(s,p,o)) => (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)
triples.partitions.length
// declare constants for BGP translation
// This constants have been generated with our encoding method
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)
// -----------------------------------------------------------
var t1 = java.lang.System.currentTimeMillis();
var pataws = triples.filter({case(s,p,o) => p==advisor}).map({case(s,p,o) => (o,s)}).
join(triples.filter({case(s,p,o) => p==worksFor}).map({case(s,p,o) => (s,o)}),part).
map({case (y,(x,z)) => (z,(x,y))}).
join(triples.filter({case(s,p,o) => p==suborg}).map({case(s,p,o) => (s,o)}),part).
map({case (y,(x,z)) => (z,(x,y))}).distinct
pataws.count
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 = triples.filter({case(s,p,o) => p==memof}).cache()
var patmsu = pmemof.map({case(s,p,o) => (o,s)}).
join(triples.filter({case(s,p,o) => p==suborg}).map({case(s,p,o) => (s,o)}),part).
map({case (y,(x,z)) => (x+""+z,(x,y,z))}).
join(triples.filter({case(s,p,o) => p==undeg}).map({case(x,p,z)=> (x+""+z,null)}), part)
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 = triples.filter({case(s,p,o) => p==advisor}).map({case(s,p,o) => (o,s)}).
join(triples.filter({case(s,p,o) => p==teaof}).map({case(s,p,o) => (s,o)}),part).
map({case (y,(x,z)) => (x+""+z,(x,y,z))}).
join(triples.filter({case(s,p,o) => p==takco}).map({case(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)
// -----------------------------------------------------------
var t1 = java.lang.System.currentTimeMillis();
var patws = triples.filter({case(s,p,o) => p==worksFor}).map({case(s,p,o) => (o,(s,p,o))}).
join(triples.filter({case(s,p,o) => p==suborg}).map({case(s,p,o) => (s,(s,p,o))}),part).
map({case (k,((s1,p1,o1),(s2,p2,o2))) => (s1,o1,o2)})
var ans_patws = patws.distinct.count
var t2= java.lang.System.currentTimeMillis();
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
import org.apache.spark.HashPartitioner
import scala.collection.mutable.ListBuffer
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");
// 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");
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")
}
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>
-encoded triples(2.1MB)
-encoded quaruples with replication(2.3MB)
-replicated quaruples(0.3MB)
-univ1.nt (16.3MB)