You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@predictionio.apache.org by sh...@apache.org on 2017/07/25 03:38:16 UTC
[04/10] incubator-predictionio git commit: [PIO-107] Removal of
examples under examples/experimental.
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/README.md
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/README.md b/examples/experimental/scala-parallel-friend-recommendation/README.md
deleted file mode 100644
index 9c94bb2..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/README.md
+++ /dev/null
@@ -1,65 +0,0 @@
-<!--
-Licensed to the Apache Software Foundation (ASF) under one or more
-contributor license agreements. See the NOTICE file distributed with
-this work for additional information regarding copyright ownership.
-The ASF licenses this file to You under the Apache License, Version 2.0
-(the "License"); you may not use this file except in compliance with
-the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
--->
-
-Parallel SimRank Algorithm
-========================================================================
-Uses the Delta-Simrank Algorithm (http://dprg.cs.uiuc.edu/docs/deltasimrank/simrank.pdf).
-
-Takes advantage of Spark's GraphX framework and in-memory/iterative computation advantage over other MapReduce frameworks such as YARN by performing each SimRank iteration as a triple of Map and two Reduce tasks.
-
-Map1 : Emit key value pairs where each key is a pair of vertex ids adjacent to the vertices with updated scores in the previous iterations, and the value is the marginal increase in SimRank (refer to the delta-simrank algorithm for exact calculation method for deltas).
-
-Reduce1 : Aggregate delta values by key.
-
-Reduce2 : Add the delta to the previous iteration SimRank score to get the current iteration's SimRank score.
-
-Prerequisite: GraphX package.
-
-Parameter Explained
--------------------
-datasource - graphEdgelistPath : The edge-list passed to GraphX's graph loader. For efficient memory storage of intermediate SimRank score calculations, the vertex ids should be in a contiguous range from 0 to (#Vertex-1). There is a utility function for re-mapping the vertex Id values : org.apache.predictionio.examples.pfriendrecommendation.DeltaSimRankRDD.normalizeGraph.
-
-The provided DataSource class uses the GraphLoader provided by GraphX. Graphs can be specified by a tab-separated edge list, where each line specifies one edge.
-The the user can refer to the provided example edge list at `$EXAMPLE_HOME/data/edge_list_small.txt` for a graph specification with 10 vertices and 20 edges.
-
-algorithms - numIterations : Number of iterations for calculating SimRank. Typical recommended is 6-8 in various papers (e.g. http://www-cs-students.stanford.edu/~glenj/simrank.pdf)
-
-algorithms - decay : Decay constant used in calculating incremental changes to SimRank
-
-### Using SimRank to Recommend Similar Items
-SimRank's intuition says two items are similar if they have similar neighbor items. In a Facebook friend recommendation scenario, neighbors to a person can be that person's friends. Two people in the same circle of friends will have higher SimRank compared to two people in different friend circles.
-For example, SimRank(Obama, George Bush) is high since they have many mutual president friends and belong to shared friend networks, whereas SimRank(Obama, Usher) is low since they have few mutual friends or friend networks.
-
-### Configurable Datasources for Sampling
-Three data sources can be configured from the engine factory :
-
-**DataSource** generates a GraphX graph using the entire dataset.
-
-**NodeSamplingDataSource** generates a GraphX graph after performing node sampling
-with induced edges between the sampled nodes. This data source takes an
-additional parameter, sampleFraction, which is the fraction of graph nodes to
-sample.
-
-**ForestFireSamplingDataSource** generates a graph after performing forest fire
-sampling. This sampling method also uses the sampleFraction parameter and takes
-an additional parameter, geoParam, which is the parameter for a geometric
-distribution that is used within forest fire sampling.
-
-### Example query
-curl -H "Content-Type: application/json" -d '{"item1":10, "item2":9}' http://localhost:8000/queries.json
-
-This queries the SimRank score between nodes with ids 10 and 9.
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/build.sbt
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/build.sbt b/examples/experimental/scala-parallel-friend-recommendation/build.sbt
deleted file mode 100644
index 57156cd..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/build.sbt
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import AssemblyKeys._
-
-assemblySettings
-
-name := "template-scala-parallel-recommendation-custom-preparator"
-
-organization := "org.apache.predictionio"
-
-libraryDependencies ++= Seq(
- "org.apache.predictionio" %% "core" % "0.9.1" % "provided",
- "org.apache.spark" %% "spark-core" % "1.2.0" % "provided",
- "org.apache.spark" %% "spark-graphx" % "1.2.0" % "provided")
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/data/circle.txt
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/data/circle.txt b/examples/experimental/scala-parallel-friend-recommendation/data/circle.txt
deleted file mode 100644
index d4f4206..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/data/circle.txt
+++ /dev/null
@@ -1,4 +0,0 @@
-1 2
-2 3
-3 1
-4 1
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/data/edge_list_small.txt
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/data/edge_list_small.txt b/examples/experimental/scala-parallel-friend-recommendation/data/edge_list_small.txt
deleted file mode 100644
index 1850ec9..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/data/edge_list_small.txt
+++ /dev/null
@@ -1,20 +0,0 @@
-0 1
-2 3
-1 3
-3 0
-6 4
-6 5
-4 1
-5 5
-7 1
-8 5
-9 5
-9 4
-10 9
-10 8
-10 5
-10 3
-10 4
-10 2
-10 1
-10 0
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/data/simple.txt
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/data/simple.txt b/examples/experimental/scala-parallel-friend-recommendation/data/simple.txt
deleted file mode 100644
index 53ed851..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/data/simple.txt
+++ /dev/null
@@ -1,4 +0,0 @@
-1 2
-1 3
-3 1
-2 3
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/engine-forest.json
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/engine-forest.json b/examples/experimental/scala-parallel-friend-recommendation/engine-forest.json
deleted file mode 100644
index 2820578..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/engine-forest.json
+++ /dev/null
@@ -1,23 +0,0 @@
-{
- "id": "SimRankParallelEngine",
- "description": "Engine for graph vertex similarity using SimRank",
- "engineFactory": "org.apache.predictionio.examples.pfriendrecommendation.PSimRankEngineFactory",
-
- "datasource": {
- "name" : "forest",
- "params": {
- "graphEdgelistPath":"data/edge_list_small.txt",
- "sampleFraction":1.0,
- "geoParam":0.7
- }
- },
- "algorithms": [
- {
- "name": "simrank",
- "params": {
- "numIterations": 6,
- "decay":0.7
- }
- }
- ]
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/engine.json
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/engine.json b/examples/experimental/scala-parallel-friend-recommendation/engine.json
deleted file mode 100644
index 413928f..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/engine.json
+++ /dev/null
@@ -1,21 +0,0 @@
-{
- "id": "SimRankParallelEngine",
- "description": "Engine for graph vertex similarity using SimRank",
- "engineFactory": "org.apache.predictionio.examples.pfriendrecommendation.PSimRankEngineFactory",
-
- "datasource": {
- "name" : "default",
- "params": {
- "graphEdgelistPath":"data/simple.txt"
- }
- },
- "algorithms": [
- {
- "name": "simrank",
- "params": {
- "numIterations": 4,
- "decay":0.7
- }
- }
- ]
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/project/assembly.sbt
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/project/assembly.sbt b/examples/experimental/scala-parallel-friend-recommendation/project/assembly.sbt
deleted file mode 100644
index 54c3252..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/project/assembly.sbt
+++ /dev/null
@@ -1 +0,0 @@
-addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.11.2")
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/DataSource.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/DataSource.scala b/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/DataSource.scala
deleted file mode 100644
index 7474180..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/DataSource.scala
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.predictionio.examples.pfriendrecommendation
-
-import org.apache.predictionio.controller.PDataSource
-import org.apache.predictionio.controller.EmptyEvaluationInfo
-import org.apache.predictionio.controller.EmptyActualResult
-import org.apache.predictionio.controller.Params
-import org.apache.predictionio.data.storage.Event
-import org.apache.predictionio.data.storage.Storage
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.graphx._
-import org.apache.spark.rdd.RDD
-
-import grizzled.slf4j.Logger
-
-/*
- Data Source Params : path to data file
- */
-case class DataSourceParams(
- val graphEdgelistPath: String
-) extends Params
-
-case class TrainingData(
- val g:Graph[Int,Int],
- val identityMatrix:RDD[(VertexId,Double)]
-)
-
-class DataSource(val dsp: DataSourceParams)
- extends PDataSource[TrainingData, EmptyEvaluationInfo, Query, Double] {
-
- override
- def readTraining(sc:SparkContext) : TrainingData = {
- val g = GraphLoader.edgeListFile(sc, dsp.graphEdgelistPath)
- // In the interest of space (since we calculate at most n*n simrank scores),
- // each of the n vertices should have vertexID in the range 0 to n-1
- // val g2 = DeltaSimRankRDD.normalizeGraph(g)
- val identity = DeltaSimRankRDD.identityMatrix(sc, g.vertices.count())
- new TrainingData(g, identity)
- }
-}
-
-case class NodeSamplingDSParams(
- val graphEdgelistPath: String,
- val sampleFraction: Double
-) extends Params
-
-class NodeSamplingDataSource(val dsp: NodeSamplingDSParams)
- extends PDataSource[TrainingData, EmptyEvaluationInfo, Query, Double] {
-
- override
- def readTraining(sc:SparkContext) : TrainingData = {
- val g = GraphLoader.edgeListFile(sc, dsp.graphEdgelistPath)
- val sampled = Sampling.nodeSampling(sc, g, dsp.sampleFraction)
- val identity = DeltaSimRankRDD.identityMatrix(sc, g.vertices.count())
- new TrainingData(sampled, identity)
- }
-}
-
-case class FFSamplingDSParams(
- val graphEdgelistPath: String,
- val sampleFraction: Double,
- val geoParam: Double
-) extends Params
-
-class ForestFireSamplingDataSource(val dsp: FFSamplingDSParams)
- extends PDataSource[TrainingData, EmptyEvaluationInfo, Query, Double] {
-
- override
- def readTraining(sc:SparkContext) : TrainingData = {
- val g = GraphLoader.edgeListFile(sc, dsp.graphEdgelistPath)
- val sampled = Sampling.forestFireSamplingInduced(
- sc,
- g,
- dsp.sampleFraction,
- dsp.geoParam)
-
- val identity = DeltaSimRankRDD.identityMatrix(sc, g.vertices.count())
- new TrainingData(sampled, identity)
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/DeltaSimRankRDD.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/DeltaSimRankRDD.scala b/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/DeltaSimRankRDD.scala
deleted file mode 100644
index 2bdde35..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/DeltaSimRankRDD.scala
+++ /dev/null
@@ -1,185 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.predictionio.examples.pfriendrecommendation
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.SparkContext._
-import org.apache.spark.graphx._
-import org.apache.spark.rdd.RDD
-
-import scala.collection.mutable.ListBuffer
-import scala.collection.mutable.Map
-
-
-object DeltaSimRankRDD {
- var decay:Double = 0.8
- var numNodes:Int = 0
-
- def calculateNthIter(
- numNodes:Int,
- g:Graph[Int, Int],
- prevDelta:RDD[((VertexId,VertexId),Double)],
- outDegreeMap:scala.collection.Map[VertexId,Long])
- : RDD[((VertexId,VertexId), Double)] =
- {
- // No changes in last iteration -> no changes this iteration.
- if (prevDelta.count() == 0)
- return prevDelta
-
- val pairList = prevDelta.toArray
- val kvPairs = pairList.map(pair => {
- val a = pair._1._1.toInt
- val b = pair._1._2.toInt
- val delta = pair._2
- val b_adj = g.edges.filter(e => e.dstId == b).map(x=>x.srcId)
- val a_adj = g.edges.filter(e => e.dstId == a).map(x=>x.srcId)
-
- val scorePairs = a_adj.cartesian(b_adj)
- scorePairs.filter(pair=> pair._1 != pair._2).map(pair => (pair, delta))
- })
-
- var union = kvPairs(0)
- var index = 0
- for (index <- 1 to kvPairs.size-1)
- union = union ++ kvPairs(index)
-
- val newDelta = union.reduceByKey(_ + _)
- .map(k =>
- (k._1, k._2*decay/(outDegreeMap(k._1._1) * outDegreeMap(k._1._2)))
- )
- newDelta
- }
-
- def identityMatrix(sc:SparkContext, numCols:Long) : RDD[(Long, Double)] =
- {
- val numElements = numCols * numCols
- val arr = Array[Long]((0L to numElements).toList:_*)
- // (Score, Index), where (x,y) = (Index/numCols, Index%numCols)
- val pairs = arr.map(x => {
- if (x/numCols == x % numCols)
- (x, 1.0)
- else
- (x, 0.0)
- })
- sc.parallelize(pairs)
- }
-
- def matrixToIndices(x:Int, y:Int, numCols:Int) = {
- x + y * numCols
- }
-
- def joinDelta(
- prevIter:RDD[(Long, Double)],
- numCols:Int,
- delta:RDD[((VertexId,VertexId), Double)]) : RDD[(Long,Double)] =
- {
- val deltaToIndex:RDD[(Long,Double)] = delta.map(x => {
- val index = (x._1._1-1)*numCols + (x._1._2-1)
- (index, x._2)
- })
- println("detaToIndex")
- deltaToIndex.foreach(println(_))
- val newIter = prevIter.leftOuterJoin(deltaToIndex)
- val newScores = newIter.map(x => {
- val index = x._1
- if (x._2._2.isDefined) {
- (index, x._2._1 + x._2._2.get)
- } else {
- (index, x._2._1)
- }
- })
- newScores
- }
-
- def getOutdegreeMap(g:Graph[Int,Int]) : scala.collection.Map[VertexId, Long] =
- {
- g.edges.map(edge => (edge.srcId,1L))
- .reduceByKey(_ + _)
- .collectAsMap()
- }
-
- def compute(
- g:Graph[Int,Int],
- numIterations:Int,
- identityMatrix:RDD[(VertexId,Double)],
- newDecay:Double) : RDD[(VertexId,Double)] =
- {
- numNodes = g.vertices.count().toInt
- decay = newDecay
-
- // Build the identity matrix representing 0-th iteration of SimRank
- val s0 = identityMatrix
- val outDegreeMap:scala.collection.Map[VertexId,Long] = getOutdegreeMap(g)
- val s0Delta = g.vertices.map(vertex => ((vertex._1, vertex._1), 1.0))
-
- var prevSimrank = s0
- var prevDelta = s0Delta
-
- println(s"initial")
- s0.foreach(println(_))
- prevDelta.foreach(println(_))
-
- for (i <- 0 to numIterations) {
- val nextIterDelta = calculateNthIter(numNodes, g, prevDelta, outDegreeMap)
- val nextIterSimrank = joinDelta(prevSimrank, numNodes, nextIterDelta)
- println(s"iteration: ${i}")
- nextIterDelta.foreach(println(_))
- nextIterSimrank.foreach(println(_))
- prevSimrank = nextIterSimrank
- prevDelta = nextIterDelta
- }
-
- prevSimrank
- }
-
- // Make all vertexId in one contiguous number range
- def normalizeGraph(g:Graph[Int,Int]) = {
- var counter = 0.toLong
- val hash = Map[VertexId, Long]()
-
- val v = g.vertices.map( pair => {
- hash(pair._1) = counter
- counter += 1
- (counter - 1, pair._2)
- })
-
- val e = g.edges.map( (e:Edge[Int]) => {
- if (hash.contains(e.srcId)) {
- e.srcId = hash(e.srcId)
- } else {
- hash += (e.srcId -> counter)
- counter += 1
- e.srcId = counter - 1
- }
-
- if (hash.contains(e.dstId)) {
- e.dstId = hash(e.dstId)
- } else {
- hash += (e.dstId -> counter)
- counter += 1
- e.dstId = counter - 1
- }
- e
- })
-
- val g2 = Graph(v,e)
- g2
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Engine.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Engine.scala b/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Engine.scala
deleted file mode 100644
index 5e0b8b9..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Engine.scala
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.predictionio.examples.pfriendrecommendation
-
-import org.apache.predictionio.controller.IEngineFactory
-import org.apache.predictionio.controller.Engine
-
-case class Query(
- val item1: Long,
- val item2: Long
-)
-
-case class PredictedResult(
- val productScores: Array[ProductScore]
-)
-
-case class ProductScore(
- product: Int,
- score: Double
-)
-
-object PSimRankEngineFactory extends IEngineFactory {
- def apply() = {
- Engine(
- Map(
- "default" -> classOf[DataSource],
- "node" -> classOf[NodeSamplingDataSource],
- "forest" -> classOf[ForestFireSamplingDataSource]),
- classOf[IdentityPreparator],
- Map("simrank" -> classOf[SimRankAlgorithm]),
- classOf[Serving])
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Preparator.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Preparator.scala b/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Preparator.scala
deleted file mode 100644
index 60b2ed5..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Preparator.scala
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.predictionio.examples.pfriendrecommendation
-
-import org.apache.predictionio.controller.EmptyParams
-import org.apache.predictionio.controller.PPreparator
-import org.apache.predictionio.controller.EmptyPreparatorParams
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.rdd.RDD
-
-import scala.io.Source // ADDED
-import org.apache.predictionio.controller.Params // ADDED
-
- // ADDED CustomPreparatorParams case class
-case class CustomPreparatorParams(
- val filepath: String
-) extends Params
-
-class IdentityPreparator(pp: EmptyPreparatorParams)
- extends PPreparator[TrainingData, TrainingData] {
-
- def prepare(sc: SparkContext, trainingData: TrainingData): TrainingData = {
- trainingData
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Sampling.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Sampling.scala b/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Sampling.scala
deleted file mode 100644
index 808780c..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Sampling.scala
+++ /dev/null
@@ -1,181 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.predictionio.examples.pfriendrecommendation
-
-import org.apache.spark.SparkContext._
-import org.apache.spark.graphx._
-import org.apache.spark.rdd.RDD
-import scala.collection.mutable.{ListBuffer, ArraySeq, HashSet}
-import scala.util.Random
-
-import scala.collection.mutable.HashSet
-import scala.collection.mutable.Queue
-import org.apache.spark.SparkContext
-import scala.collection.mutable.Map
-
-object Sampling {
-
- def geometricSample(param: Double) : Int = {
- var num = 1
- while(Random.nextDouble <= param) {
- num += 1
- }
- num
- }
-
- def sortBySrc(a:Array[Edge[Int]]): Array[Edge[Int]] = {
- if (a.length < 2) {
- a
- } else {
- val pivot = a(a.length / 2).srcId
- // 'L'ess, 'E'qual, 'G'reater
- val partitions = a.groupBy( (e:Edge[Int]) => {
- if (e.srcId < pivot)
- 'L'
- else if (e.srcId > pivot)
- 'G'
- else
- 'E'
- })
-
- var sortedAccumulator: Array[Edge[Int]] = Array()
- List('L', 'E', 'G').foreach((c:Char) => {
- if (partitions.contains(c)) {
- sortedAccumulator = sortedAccumulator ++ partitions(c)
- }
- })
- sortedAccumulator
- }
- }
-
- // Samples vertices by forest fire random process and induces edges.
- // Fraction denotes fraction of total graph vertices to sample and geoParam
- // denotes the parameter for geometric distribution, which is used to
- // determine branching factor at each iteration of forest fire process.
- def forestFireSamplingInduced (
- sc: SparkContext,
- graph: Graph[Int,Int],
- fraction: Double,
- geoParam: Double = 0.7) =
- {
- var g = graph
- var e = sortBySrc(g.edges.toArray)
- val targetVertexCount = (graph.vertices.count() * fraction).toInt
- var seedVertices = graph.vertices
- .sample(false, fraction, targetVertexCount)
- .toArray.iterator
- var sampledVertices: HashSet[VertexId] = HashSet()
- var burnQueue: Queue[VertexId] = Queue()
-
- while (sampledVertices.size < targetVertexCount) {
- val seedVertex = seedVertices.next
- sampledVertices += seedVertex._1
- burnQueue += seedVertex._1
- while (burnQueue.size > 0 ){
- val vertexId = burnQueue.dequeue()
- val numToSample = geometricSample(geoParam)
- val edgeCandidates = accumulateEdges(e, vertexId)
- val burnCandidate = sc.parallelize(edgeCandidates)
- .filter((e:Edge[Int]) => {
- !sampledVertices.contains(e.dstId)
- })
- val burnFraction = numToSample.toDouble / burnCandidate.count.toDouble
- val burnEdges = burnCandidate.sample(
- false,
- burnFraction,
- Random.nextLong)
- val neighborVertexIds = burnEdges.map((e:Edge[Int]) => e.dstId)
- sampledVertices = sampledVertices ++ neighborVertexIds.toArray
- burnQueue = burnQueue ++ neighborVertexIds.toArray
-
- if (sampledVertices.size > targetVertexCount) {
- burnQueue.dequeueAll((v:VertexId) => true)
- }
- }
- }
- val vertex: Seq[(VertexId, Int)] = sampledVertices.map(v => (v,1))
- .toSeq
- val edges = graph.edges.filter(e =>
- sampledVertices.contains(e.srcId) && sampledVertices.contains(e.dstId)
- )
- Graph(sc.parallelize(vertex), edges)
- }
-
- // Samples vertices uniformly and induces edges.
- def nodeSampling(sc:SparkContext, graph:Graph[Int,Int], fraction:Double) = {
- val vertices = graph.vertices.sample(false, fraction, Random.nextLong)
- val vertexMap = vertices.collectAsMap()
- val edges = graph.edges.filter(e =>
- vertexMap.contains(e.srcId) && vertexMap.contains(e.dstId)
- )
- val graph2 = Graph(vertices, edges)
- graph2
- }
-
- // Get all edges with source vertexId of target
- def accumulateEdges(
- e:Array[Edge[Int]],
- target:VertexId) : ListBuffer[Edge[Int]] =
- {
- val idx = binarySearchE(e, target)(0, e.size-1)
- var outEdges: ListBuffer[Edge[Int]] = ListBuffer()
- if (idx == -1)
- return outEdges
- outEdges.append(e(idx))
- var tIdx = idx+1
- var edge:Edge[Int] = null
- // get upper edges
- while (tIdx < e.size) {
- edge = e(tIdx)
- if (edge.srcId == target) {
- outEdges.append(edge)
- tIdx += 1
- } else {
- tIdx = e.size
- }
- }
- // get lower edges
- tIdx = idx-1
- while (tIdx > -1){
- edge = e(tIdx)
- if (edge.srcId == target) {
- outEdges.append(edge)
- tIdx -= 1
- } else {
- tIdx = -1
- }
- }
- outEdges
- }
-
-
- // Binary search to find an edge with target vertexId
- def binarySearchE(list: Array[Edge[Int]], target: VertexId)
- (start: Int=0, end: Int=list.length-1): Int =
- {
- if (start>end)
- return -1
- val mid = start + (end-start+1)/2
- if (list(mid).srcId == target)
- return mid
- else if (list(mid).srcId > target)
- return binarySearchE(list, target)(start, mid-1)
- else
- return binarySearchE(list, target)(mid+1, end)
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Serving.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Serving.scala b/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Serving.scala
deleted file mode 100644
index eca5d2d..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/Serving.scala
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.predictionio.examples.pfriendrecommendation
-
-import org.apache.predictionio.controller.LServing
-import org.apache.predictionio.controller.EmptyServingParams
-
-class Serving
- extends LServing[Query, Double] {
-
- override
- def serve(query: Query, predictedResults: Seq[Double]): Double = {
- predictedResults.head
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/SimRankAlgorithm.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/SimRankAlgorithm.scala b/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/SimRankAlgorithm.scala
deleted file mode 100644
index cd620ba..0000000
--- a/examples/experimental/scala-parallel-friend-recommendation/src/main/scala/SimRankAlgorithm.scala
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.predictionio.examples.pfriendrecommendation
-import org.apache.predictionio.controller.PAlgorithm
-import org.apache.predictionio.controller.Params
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.graphx._
-import org.apache.spark.rdd.RDD
-
-case class SimRankParams(
- val numIterations: Int,
- val decay: Double) extends Params
-
-class SimRankAlgorithm(val ap: SimRankParams)
- extends PAlgorithm[TrainingData, RDD[(Long,Double)], Query, Double] {
-
- def train(td: TrainingData): RDD[(Long,Double)] = {
- td.g.edges.count()
- val scores = DeltaSimRankRDD.compute(
- td.g,
- ap.numIterations,
- td.identityMatrix,
- ap.decay)
- scores
- }
-
- /*
- def batchPredict(
- model: RDD[(VertexId,Double)],
- feature: RDD[(Long, (Int, Int))]): RDD[(Long, Double)] = {
- feature.map(x => (x._1, predict(model, (x._2._1, x._2._1))))
- }
- */
-
- def predict(
- model: RDD[(Long,Double)], query:Query): Double = {
- // Will never encounter rounding errors because model is an n*n "matrix".
- val numElems = math.sqrt(model.count()).toInt
- val index = query.item1 * numElems + query.item2
- val seq = model.lookup(index)
- seq.head
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-cat/.gitignore
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-cat/.gitignore b/examples/experimental/scala-parallel-recommendation-cat/.gitignore
deleted file mode 100644
index 64fa18b..0000000
--- a/examples/experimental/scala-parallel-recommendation-cat/.gitignore
+++ /dev/null
@@ -1,3 +0,0 @@
-manifest.json
-target/
-pio.log
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-cat/README.md
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-cat/README.md b/examples/experimental/scala-parallel-recommendation-cat/README.md
deleted file mode 100644
index 9394dbf..0000000
--- a/examples/experimental/scala-parallel-recommendation-cat/README.md
+++ /dev/null
@@ -1,94 +0,0 @@
-<!--
-Licensed to the Apache Software Foundation (ASF) under one or more
-contributor license agreements. See the NOTICE file distributed with
-this work for additional information regarding copyright ownership.
-The ASF licenses this file to You under the Apache License, Version 2.0
-(the "License"); you may not use this file except in compliance with
-the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
--->
-
-# Recommendation Template support categories filtering
-
-## Documentation
-
-The data requirement is similar to "Similar Product Template". Please see.
-http://predictionio.incubator.apache.org/templates/similarproduct/quickstart/
-
-By default, view events and MLlib ALS trainImplicit is used.
-
-The main difference is:
-
-Query by user Id instead of list of item Ids.
-
-## Development Notes
-
-### import sample data
-
-```
-$ python data/import_eventserver.py --access_key <your_access_key>
-```
-
-### query
-
-normal:
-
-```
-$ curl -H "Content-Type: application/json" \
--d '{
- "user" : "u1",
- "num" : 10 }' \
-http://localhost:8000/queries.json \
--w %{time_connect}:%{time_starttransfer}:%{time_total}
-```
-
-```
-$ curl -H "Content-Type: application/json" \
--d '{
- "user" : "u1",
- "num": 10,
- "categories" : ["c4", "c3"]
-}' \
-http://localhost:8000/queries.json \
--w %{time_connect}:%{time_starttransfer}:%{time_total}
-```
-
-```
-curl -H "Content-Type: application/json" \
--d '{
- "user" : "u1",
- "num": 10,
- "whiteList": ["i21", "i26", "i40"]
-}' \
-http://localhost:8000/queries.json \
--w %{time_connect}:%{time_starttransfer}:%{time_total}
-```
-
-```
-curl -H "Content-Type: application/json" \
--d '{
- "user" : "u1",
- "num": 10,
- "blackList": ["i21", "i26", "i40"]
-}' \
-http://localhost:8000/queries.json \
--w %{time_connect}:%{time_starttransfer}:%{time_total}
-```
-
-unknown user:
-
-```
-curl -H "Content-Type: application/json" \
--d '{
- "user" : "unk1",
- "num": 10}' \
-http://localhost:8000/queries.json \
--w %{time_connect}:%{time_starttransfer}:%{time_total}
-```
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-cat/build.sbt
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-cat/build.sbt b/examples/experimental/scala-parallel-recommendation-cat/build.sbt
deleted file mode 100644
index 3ff8d27..0000000
--- a/examples/experimental/scala-parallel-recommendation-cat/build.sbt
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import AssemblyKeys._
-
-assemblySettings
-
-name := "template-scala-parallel-recommendation-cat"
-
-organization := "org.apache.predictionio"
-
-libraryDependencies ++= Seq(
- "org.apache.predictionio" %% "core" % "0.8.6" % "provided",
- "org.apache.spark" %% "spark-core" % "1.2.0" % "provided",
- "org.apache.spark" %% "spark-mllib" % "1.2.0" % "provided")
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-cat/data/import_eventserver.py
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-cat/data/import_eventserver.py b/examples/experimental/scala-parallel-recommendation-cat/data/import_eventserver.py
deleted file mode 100644
index d47d60f..0000000
--- a/examples/experimental/scala-parallel-recommendation-cat/data/import_eventserver.py
+++ /dev/null
@@ -1,101 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-"""
-Import sample data for E-Commerce Recommendation Engine Template
-"""
-
-import predictionio
-import argparse
-import random
-
-SEED = 3
-
-def import_events(client):
- random.seed(SEED)
- count = 0
- print client.get_status()
- print "Importing data..."
-
- # generate 10 users, with user ids u1,u2,....,u10
- user_ids = ["u%s" % i for i in range(1, 11)]
- for user_id in user_ids:
- print "Set user", user_id
- client.create_event(
- event="$set",
- entity_type="user",
- entity_id=user_id
- )
- count += 1
-
- # generate 50 items, with item ids i1,i2,....,i50
- # random assign 1 to 4 categories among c1-c6 to items
- categories = ["c%s" % i for i in range(1, 7)]
- item_ids = ["i%s" % i for i in range(1, 51)]
- for item_id in item_ids:
- print "Set item", item_id
- client.create_event(
- event="$set",
- entity_type="item",
- entity_id=item_id,
- properties={
- "categories" : random.sample(categories, random.randint(1, 4))
- }
- )
- count += 1
-
- # each user randomly viewed 10 items
- for user_id in user_ids:
- for viewed_item in random.sample(item_ids, 10):
- print "User", user_id ,"views item", viewed_item
- client.create_event(
- event="view",
- entity_type="user",
- entity_id=user_id,
- target_entity_type="item",
- target_entity_id=viewed_item
- )
- count += 1
- # randomly buy some of the viewed items
- if random.choice([True, False]):
- print "User", user_id ,"buys item", viewed_item
- client.create_event(
- event="buy",
- entity_type="user",
- entity_id=user_id,
- target_entity_type="item",
- target_entity_id=viewed_item
- )
- count += 1
-
- print "%s events are imported." % count
-
-if __name__ == '__main__':
- parser = argparse.ArgumentParser(
- description="Import sample data for e-commerce recommendation engine")
- parser.add_argument('--access_key', default='invald_access_key')
- parser.add_argument('--url', default="http://localhost:7070")
-
- args = parser.parse_args()
- print args
-
- client = predictionio.EventClient(
- access_key=args.access_key,
- url=args.url,
- threads=5,
- qsize=500)
- import_events(client)
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-cat/data/send_query.py
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-cat/data/send_query.py b/examples/experimental/scala-parallel-recommendation-cat/data/send_query.py
deleted file mode 100644
index 21e9146..0000000
--- a/examples/experimental/scala-parallel-recommendation-cat/data/send_query.py
+++ /dev/null
@@ -1,24 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-"""
-Send sample query to prediction engine
-"""
-
-import predictionio
-engine_client = predictionio.EngineClient(url="http://localhost:8000")
-print engine_client.send_query({"user": "u1", "num": 4})
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-cat/engine.json
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-cat/engine.json b/examples/experimental/scala-parallel-recommendation-cat/engine.json
deleted file mode 100644
index d3406af..0000000
--- a/examples/experimental/scala-parallel-recommendation-cat/engine.json
+++ /dev/null
@@ -1,21 +0,0 @@
-{
- "id": "default",
- "description": "Default settings",
- "engineFactory": "org.template.recommendation.RecommendationEngine",
- "datasource": {
- "params" : {
- "appId": 17
- }
- },
- "algorithms": [
- {
- "name": "als",
- "params": {
- "rank": 10,
- "numIterations" : 20,
- "lambda": 0.01,
- "seed": 3
- }
- }
- ]
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-cat/project/assembly.sbt
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-cat/project/assembly.sbt b/examples/experimental/scala-parallel-recommendation-cat/project/assembly.sbt
deleted file mode 100644
index 54c3252..0000000
--- a/examples/experimental/scala-parallel-recommendation-cat/project/assembly.sbt
+++ /dev/null
@@ -1 +0,0 @@
-addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.11.2")
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/ALSAlgorithm.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/ALSAlgorithm.scala b/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/ALSAlgorithm.scala
deleted file mode 100644
index 84ed3a4..0000000
--- a/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/ALSAlgorithm.scala
+++ /dev/null
@@ -1,279 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.template.recommendation
-
-import org.apache.predictionio.controller.P2LAlgorithm
-import org.apache.predictionio.controller.Params
-import org.apache.predictionio.data.storage.BiMap
-import org.apache.predictionio.data.storage.Event
-import org.apache.predictionio.data.storage.Storage
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.mllib.recommendation.ALS
-import org.apache.spark.mllib.recommendation.{Rating => MLlibRating}
-
-import grizzled.slf4j.Logger
-
-import scala.collection.mutable.PriorityQueue
-import scala.concurrent.duration.Duration
-import scala.concurrent.ExecutionContext.Implicits.global
-
-case class ALSAlgorithmParams(
- rank: Int,
- numIterations: Int,
- lambda: Double,
- seed: Option[Long]
-) extends Params
-
-class ALSModel(
- val rank: Int,
- val userFeatures: Map[Int, Array[Double]],
- val productFeatures: Map[Int, (Item, Option[Array[Double]])],
- val userStringIntMap: BiMap[String, Int],
- val itemStringIntMap: BiMap[String, Int]
-) extends Serializable {
-
- @transient lazy val itemIntStringMap = itemStringIntMap.inverse
-
- override def toString = {
- s" rank: ${rank}" +
- s" userFeatures: [${userFeatures.size}]" +
- s"(${userFeatures.take(2).toList}...)" +
- s" productFeatures: [${productFeatures.size}]" +
- s"(${productFeatures.take(2).toList}...)" +
- s" userStringIntMap: [${userStringIntMap.size}]" +
- s"(${userStringIntMap.take(2).toString}...)]" +
- s" itemStringIntMap: [${itemStringIntMap.size}]" +
- s"(${itemStringIntMap.take(2).toString}...)]"
- }
-}
-
-/**
- * Use ALS to build item x feature matrix
- */
-class ALSAlgorithm(val ap: ALSAlgorithmParams)
- extends P2LAlgorithm[PreparedData, ALSModel, Query, PredictedResult] {
-
- @transient lazy val logger = Logger[this.type]
-
- def train(data: PreparedData): ALSModel = {
- require(!data.viewEvents.take(1).isEmpty,
- s"viewEvents in PreparedData cannot be empty." +
- " Please check if DataSource generates TrainingData" +
- " and Preprator generates PreparedData correctly.")
- require(!data.users.take(1).isEmpty,
- s"users in PreparedData cannot be empty." +
- " Please check if DataSource generates TrainingData" +
- " and Preprator generates PreparedData correctly.")
- require(!data.items.take(1).isEmpty,
- s"items in PreparedData cannot be empty." +
- " Please check if DataSource generates TrainingData" +
- " and Preprator generates PreparedData correctly.")
- // create User and item's String ID to integer index BiMap
- val userStringIntMap = BiMap.stringInt(data.users.keys)
- val itemStringIntMap = BiMap.stringInt(data.items.keys)
-
- val mllibRatings = data.viewEvents
- .map { r =>
- // Convert user and item String IDs to Int index for MLlib
- val uindex = userStringIntMap.getOrElse(r.user, -1)
- val iindex = itemStringIntMap.getOrElse(r.item, -1)
-
- if (uindex == -1)
- logger.info(s"Couldn't convert nonexistent user ID ${r.user}"
- + " to Int index.")
-
- if (iindex == -1)
- logger.info(s"Couldn't convert nonexistent item ID ${r.item}"
- + " to Int index.")
-
- ((uindex, iindex), 1)
- }.filter { case ((u, i), v) =>
- // keep events with valid user and item index
- (u != -1) && (i != -1)
- }
- .reduceByKey(_ + _) // aggregate all view events of same user-item pair
- .map { case ((u, i), v) =>
- // MLlibRating requires integer index for user and item
- MLlibRating(u, i, v)
- }.cache()
-
- // MLLib ALS cannot handle empty training data.
- require(!mllibRatings.take(1).isEmpty,
- s"mllibRatings cannot be empty." +
- " Please check if your events contain valid user and item ID.")
-
- // seed for MLlib ALS
- val seed = ap.seed.getOrElse(System.nanoTime)
-
- val m = ALS.trainImplicit(
- ratings = mllibRatings,
- rank = ap.rank,
- iterations = ap.numIterations,
- lambda = ap.lambda,
- blocks = -1,
- alpha = 1.0,
- seed = seed)
-
- val userFeatures = m.userFeatures.collectAsMap.toMap
-
- // convert ID to Int index
- val items = data.items.map { case (id, item) =>
- (itemStringIntMap(id), item)
- }
-
- // join item with the trained productFeatures
- val productFeatures = items.leftOuterJoin(m.productFeatures)
- .collectAsMap.toMap
-
- new ALSModel(
- rank = m.rank,
- userFeatures = userFeatures,
- productFeatures = productFeatures,
- userStringIntMap = userStringIntMap,
- itemStringIntMap = itemStringIntMap
- )
- }
-
- def predict(model: ALSModel, query: Query): PredictedResult = {
-
- val userFeatures = model.userFeatures
- val productFeatures = model.productFeatures
-
- // convert whiteList's string ID to integer index
- val whiteList: Option[Set[Int]] = query.whiteList.map( set =>
- set.map(model.itemStringIntMap.get(_)).flatten
- )
-
- val blackList: Set[String] = query.blackList.getOrElse(Set[String]())
-
- // combine query's blackList
- // into final blackList.
- // convert seen Items list from String ID to interger Index
- val finalBlackList: Set[Int] = blackList.map( x =>
- model.itemStringIntMap.get(x)).flatten
-
- val userFeature =
- model.userStringIntMap.get(query.user).map { userIndex =>
- userFeatures.get(userIndex)
- }
- // flatten Option[Option[Array[Double]]] to Option[Array[Double]]
- .flatten
-
- val topScores = if (userFeature.isDefined) {
- // the user has feature vector
- val uf = userFeature.get
- val indexScores: Map[Int, Double] =
- productFeatures.par // convert to parallel collection
- .filter { case (i, (item, feature)) =>
- feature.isDefined &&
- isCandidateItem(
- i = i,
- item = item,
- categories = query.categories,
- whiteList = whiteList,
- blackList = finalBlackList
- )
- }
- .map { case (i, (item, feature)) =>
- // NOTE: feature must be defined, so can call .get
- val s = dotProduct(uf, feature.get)
- // Can adjust score here
- (i, s)
- }
- .filter(_._2 > 0) // only keep items with score > 0
- .seq // convert back to sequential collection
-
- val ord = Ordering.by[(Int, Double), Double](_._2).reverse
- val topScores = getTopN(indexScores, query.num)(ord).toArray
-
- topScores
-
- } else {
- // the user doesn't have feature vector.
- // For example, new user is created after model is trained.
- logger.info(s"No userFeature found for user ${query.user}.")
- Array[(Int, Double)]()
- }
-
- val itemScores = topScores.map { case (i, s) =>
- new ItemScore(
- // convert item int index back to string ID
- item = model.itemIntStringMap(i),
- score = s
- )
- }
-
- new PredictedResult(itemScores)
- }
-
- private
- def getTopN[T](s: Iterable[T], n: Int)(implicit ord: Ordering[T]): Seq[T] = {
-
- val q = PriorityQueue()
-
- for (x <- s) {
- if (q.size < n)
- q.enqueue(x)
- else {
- // q is full
- if (ord.compare(x, q.head) < 0) {
- q.dequeue()
- q.enqueue(x)
- }
- }
- }
-
- q.dequeueAll.toSeq.reverse
- }
-
- private
- def dotProduct(v1: Array[Double], v2: Array[Double]): Double = {
- val size = v1.size
- var i = 0
- var d: Double = 0
- while (i < size) {
- d += v1(i) * v2(i)
- i += 1
- }
- d
- }
-
- private
- def isCandidateItem(
- i: Int,
- item: Item,
- categories: Option[Set[String]],
- whiteList: Option[Set[Int]],
- blackList: Set[Int]
- ): Boolean = {
- // can add other custom filtering here
- whiteList.map(_.contains(i)).getOrElse(true) &&
- !blackList.contains(i) &&
- // filter categories
- categories.map { cat =>
- item.categories.map { itemCat =>
- // keep this item if has ovelap categories with the query
- !(itemCat.toSet.intersect(cat).isEmpty)
- }.getOrElse(false) // discard this item if it has no categories
- }.getOrElse(true)
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/DataSource.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/DataSource.scala b/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/DataSource.scala
deleted file mode 100644
index a7e8f5b..0000000
--- a/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/DataSource.scala
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.template.recommendation
-
-import org.apache.predictionio.controller.PDataSource
-import org.apache.predictionio.controller.EmptyEvaluationInfo
-import org.apache.predictionio.controller.EmptyActualResult
-import org.apache.predictionio.controller.Params
-import org.apache.predictionio.data.storage.Event
-import org.apache.predictionio.data.storage.Storage
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.rdd.RDD
-
-import grizzled.slf4j.Logger
-
-case class DataSourceParams(appId: Int) extends Params
-
-class DataSource(val dsp: DataSourceParams)
- extends PDataSource[TrainingData,
- EmptyEvaluationInfo, Query, EmptyActualResult] {
-
- @transient lazy val logger = Logger[this.type]
-
- override
- def readTraining(sc: SparkContext): TrainingData = {
- val eventsDb = Storage.getPEvents()
-
- // create a RDD of (entityID, User)
- val usersRDD: RDD[(String, User)] = eventsDb.aggregateProperties(
- appId = dsp.appId,
- entityType = "user"
- )(sc).map { case (entityId, properties) =>
- val user = try {
- User()
- } catch {
- case e: Exception => {
- logger.error(s"Failed to get properties ${properties} of" +
- s" user ${entityId}. Exception: ${e}.")
- throw e
- }
- }
- (entityId, user)
- }.cache()
-
- // create a RDD of (entityID, Item)
- val itemsRDD: RDD[(String, Item)] = eventsDb.aggregateProperties(
- appId = dsp.appId,
- entityType = "item"
- )(sc).map { case (entityId, properties) =>
- val item = try {
- // Assume categories is optional property of item.
- Item(categories = properties.getOpt[List[String]]("categories"))
- } catch {
- case e: Exception => {
- logger.error(s"Failed to get properties ${properties} of" +
- s" item ${entityId}. Exception: ${e}.")
- throw e
- }
- }
- (entityId, item)
- }.cache()
-
- // get all "user" "view" "item" events
- val viewEventsRDD: RDD[ViewEvent] = eventsDb.find(
- appId = dsp.appId,
- entityType = Some("user"),
- eventNames = Some(List("view")),
- // targetEntityType is optional field of an event.
- targetEntityType = Some(Some("item")))(sc)
- // eventsDb.find() returns RDD[Event]
- .map { event =>
- val viewEvent = try {
- event.event match {
- case "view" => ViewEvent(
- user = event.entityId,
- item = event.targetEntityId.get,
- t = event.eventTime.getMillis)
- case _ => throw new Exception(s"Unexpected event ${event} is read.")
- }
- } catch {
- case e: Exception => {
- logger.error(s"Cannot convert ${event} to ViewEvent." +
- s" Exception: ${e}.")
- throw e
- }
- }
- viewEvent
- }.cache()
-
- new TrainingData(
- users = usersRDD,
- items = itemsRDD,
- viewEvents = viewEventsRDD
- )
- }
-}
-
-case class User()
-
-case class Item(categories: Option[List[String]])
-
-case class ViewEvent(user: String, item: String, t: Long)
-
-class TrainingData(
- val users: RDD[(String, User)],
- val items: RDD[(String, Item)],
- val viewEvents: RDD[ViewEvent]
-) extends Serializable {
- override def toString = {
- s"users: [${users.count()} (${users.take(2).toList}...)]" +
- s"items: [${items.count()} (${items.take(2).toList}...)]" +
- s"viewEvents: [${viewEvents.count()}] (${viewEvents.take(2).toList}...)"
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/Engine.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/Engine.scala b/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/Engine.scala
deleted file mode 100644
index 32e2760..0000000
--- a/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/Engine.scala
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.template.recommendation
-
-import org.apache.predictionio.controller.IEngineFactory
-import org.apache.predictionio.controller.Engine
-
-case class Query(
- user: String,
- num: Int,
- categories: Option[Set[String]],
- whiteList: Option[Set[String]],
- blackList: Option[Set[String]]
-)
-
-case class PredictedResult(
- itemScores: Array[ItemScore]
-)
-
-case class ItemScore(
- item: String,
- score: Double
-)
-
-object RecommendationEngine extends IEngineFactory {
- def apply() = {
- new Engine(
- classOf[DataSource],
- classOf[Preparator],
- Map("als" -> classOf[ALSAlgorithm]),
- classOf[Serving])
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/Preparator.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/Preparator.scala b/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/Preparator.scala
deleted file mode 100644
index 3193317..0000000
--- a/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/Preparator.scala
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.template.recommendation
-
-import org.apache.predictionio.controller.PPreparator
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.rdd.RDD
-
-class Preparator
- extends PPreparator[TrainingData, PreparedData] {
-
- def prepare(sc: SparkContext, trainingData: TrainingData): PreparedData = {
- new PreparedData(
- users = trainingData.users,
- items = trainingData.items,
- viewEvents = trainingData.viewEvents)
- }
-}
-
-class PreparedData(
- val users: RDD[(String, User)],
- val items: RDD[(String, Item)],
- val viewEvents: RDD[ViewEvent]
-) extends Serializable
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/Serving.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/Serving.scala b/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/Serving.scala
deleted file mode 100644
index 48fda35..0000000
--- a/examples/experimental/scala-parallel-recommendation-cat/src/main/scala/Serving.scala
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.template.recommendation
-
-import org.apache.predictionio.controller.LServing
-
-class Serving
- extends LServing[Query, PredictedResult] {
-
- override def serve(query: Query,
- predictedResults: Seq[PredictedResult]): PredictedResult = {
- predictedResults.head
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-custom-datasource/.gitignore
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-custom-datasource/.gitignore b/examples/experimental/scala-parallel-recommendation-custom-datasource/.gitignore
deleted file mode 100644
index ea4e89d..0000000
--- a/examples/experimental/scala-parallel-recommendation-custom-datasource/.gitignore
+++ /dev/null
@@ -1,2 +0,0 @@
-data/sample_movielens_data.txt
-manifest.json
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-custom-datasource/build.sbt
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-custom-datasource/build.sbt b/examples/experimental/scala-parallel-recommendation-custom-datasource/build.sbt
deleted file mode 100644
index 91e94f2..0000000
--- a/examples/experimental/scala-parallel-recommendation-custom-datasource/build.sbt
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import AssemblyKeys._
-
-assemblySettings
-
-name := "template-scala-parallel-recommendation"
-
-organization := "org.apache.predictionio"
-
-libraryDependencies ++= Seq(
- "org.apache.predictionio" %% "core" % "0.9.1" % "provided",
- "org.apache.spark" %% "spark-core" % "1.2.0" % "provided",
- "org.apache.spark" %% "spark-mllib" % "1.2.0" % "provided")
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-custom-datasource/data/send_query.py
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-custom-datasource/data/send_query.py b/examples/experimental/scala-parallel-recommendation-custom-datasource/data/send_query.py
deleted file mode 100644
index ca19dc5..0000000
--- a/examples/experimental/scala-parallel-recommendation-custom-datasource/data/send_query.py
+++ /dev/null
@@ -1,24 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-"""
-Send sample query to prediction engine
-"""
-
-import predictionio
-engine_client = predictionio.EngineClient(url="http://localhost:8000")
-print engine_client.send_query({"user": "1", "num": 4})
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-custom-datasource/engine.json
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-custom-datasource/engine.json b/examples/experimental/scala-parallel-recommendation-custom-datasource/engine.json
deleted file mode 100644
index 4f24334..0000000
--- a/examples/experimental/scala-parallel-recommendation-custom-datasource/engine.json
+++ /dev/null
@@ -1,20 +0,0 @@
-{
- "id": "default",
- "description": "Default settings",
- "engineFactory": "org.template.recommendation.RecommendationEngine",
- "datasource": {
- "params": {
- "filepath": "./data/sample_movielens_data.txt"
- }
- },
- "algorithms": [
- {
- "name": "als",
- "params": {
- "rank": 10,
- "numIterations": 20,
- "lambda": 0.01
- }
- }
- ]
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-custom-datasource/project/assembly.sbt
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-custom-datasource/project/assembly.sbt b/examples/experimental/scala-parallel-recommendation-custom-datasource/project/assembly.sbt
deleted file mode 100644
index 54c3252..0000000
--- a/examples/experimental/scala-parallel-recommendation-custom-datasource/project/assembly.sbt
+++ /dev/null
@@ -1 +0,0 @@
-addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.11.2")
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/ALSAlgorithm.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/ALSAlgorithm.scala b/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/ALSAlgorithm.scala
deleted file mode 100644
index fd93407..0000000
--- a/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/ALSAlgorithm.scala
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.template.recommendation
-
-import org.apache.predictionio.controller.PAlgorithm
-import org.apache.predictionio.controller.Params
-import org.apache.predictionio.data.storage.BiMap
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.rdd.RDD
-import org.apache.spark.mllib.recommendation.ALS
-import org.apache.spark.mllib.recommendation.{Rating => MLlibRating}
-import org.apache.spark.mllib.recommendation.ALSModel
-
-import grizzled.slf4j.Logger
-
-case class ALSAlgorithmParams(
- rank: Int,
- numIterations: Int,
- lambda: Double) extends Params
-
-class ALSAlgorithm(val ap: ALSAlgorithmParams)
- extends PAlgorithm[PreparedData, ALSModel, Query, PredictedResult] {
-
- @transient lazy val logger = Logger[this.type]
-
- def train(data: PreparedData): ALSModel = {
- // Convert user and item String IDs to Int index for MLlib
- val userStringIntMap = BiMap.stringInt(data.ratings.map(_.user))
- val itemStringIntMap = BiMap.stringInt(data.ratings.map(_.item))
- val mllibRatings = data.ratings.map( r =>
- // MLlibRating requires integer index for user and item
- MLlibRating(userStringIntMap(r.user), itemStringIntMap(r.item), r.rating)
- )
- val m = ALS.train(mllibRatings, ap.rank, ap.numIterations, ap.lambda)
- new ALSModel(
- rank = m.rank,
- userFeatures = m.userFeatures,
- productFeatures = m.productFeatures,
- userStringIntMap = userStringIntMap,
- itemStringIntMap = itemStringIntMap)
- }
-
- def predict(model: ALSModel, query: Query): PredictedResult = {
- // Convert String ID to Int index for Mllib
- model.userStringIntMap.get(query.user).map { userInt =>
- // create inverse view of itemStringIntMap
- val itemIntStringMap = model.itemStringIntMap.inverse
- // recommendProducts() returns Array[MLlibRating], which uses item Int
- // index. Convert it to String ID for returning PredictedResult
- val itemScores = model.recommendProducts(userInt, query.num)
- .map (r => ItemScore(itemIntStringMap(r.product), r.rating))
- new PredictedResult(itemScores)
- }.getOrElse{
- logger.info(s"No prediction for unknown user ${query.user}.")
- new PredictedResult(Array.empty)
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/ALSModel.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/ALSModel.scala b/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/ALSModel.scala
deleted file mode 100644
index 4697732..0000000
--- a/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/ALSModel.scala
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.mllib.recommendation
-// This must be the same package as Spark's MatrixFactorizationModel because
-// MatrixFactorizationModel's constructor is private and we are using
-// its constructor in order to save and load the model
-
-import org.template.recommendation.ALSAlgorithmParams
-
-import org.apache.predictionio.controller.IPersistentModel
-import org.apache.predictionio.controller.IPersistentModelLoader
-import org.apache.predictionio.data.storage.BiMap
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.rdd.RDD
-
-class ALSModel(
- override val rank: Int,
- override val userFeatures: RDD[(Int, Array[Double])],
- override val productFeatures: RDD[(Int, Array[Double])],
- val userStringIntMap: BiMap[String, Int],
- val itemStringIntMap: BiMap[String, Int])
- extends MatrixFactorizationModel(rank, userFeatures, productFeatures)
- with IPersistentModel[ALSAlgorithmParams] {
-
- def save(id: String, params: ALSAlgorithmParams,
- sc: SparkContext): Boolean = {
-
- sc.parallelize(Seq(rank)).saveAsObjectFile(s"/tmp/${id}/rank")
- userFeatures.saveAsObjectFile(s"/tmp/${id}/userFeatures")
- productFeatures.saveAsObjectFile(s"/tmp/${id}/productFeatures")
- sc.parallelize(Seq(userStringIntMap))
- .saveAsObjectFile(s"/tmp/${id}/userStringIntMap")
- sc.parallelize(Seq(itemStringIntMap))
- .saveAsObjectFile(s"/tmp/${id}/itemStringIntMap")
- true
- }
-
- override def toString = {
- s"userFeatures: [${userFeatures.count()}]" +
- s"(${userFeatures.take(2).toList}...)" +
- s" productFeatures: [${productFeatures.count()}]" +
- s"(${productFeatures.take(2).toList}...)" +
- s" userStringIntMap: [${userStringIntMap.size}]" +
- s"(${userStringIntMap.take(2)}...)" +
- s" itemStringIntMap: [${itemStringIntMap.size}]" +
- s"(${itemStringIntMap.take(2)}...)"
- }
-}
-
-object ALSModel
- extends IPersistentModelLoader[ALSAlgorithmParams, ALSModel] {
- def apply(id: String, params: ALSAlgorithmParams,
- sc: Option[SparkContext]) = {
- new ALSModel(
- rank = sc.get.objectFile[Int](s"/tmp/${id}/rank").first,
- userFeatures = sc.get.objectFile(s"/tmp/${id}/userFeatures"),
- productFeatures = sc.get.objectFile(s"/tmp/${id}/productFeatures"),
- userStringIntMap = sc.get
- .objectFile[BiMap[String, Int]](s"/tmp/${id}/userStringIntMap").first,
- itemStringIntMap = sc.get
- .objectFile[BiMap[String, Int]](s"/tmp/${id}/itemStringIntMap").first)
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/DataSource.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/DataSource.scala b/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/DataSource.scala
deleted file mode 100644
index fed37a9..0000000
--- a/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/DataSource.scala
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.template.recommendation
-
-import org.apache.predictionio.controller.PDataSource
-import org.apache.predictionio.controller.EmptyEvaluationInfo
-import org.apache.predictionio.controller.EmptyActualResult
-import org.apache.predictionio.controller.Params
-import org.apache.predictionio.data.storage.Event
-import org.apache.predictionio.data.storage.Storage
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.rdd.RDD
-
-import grizzled.slf4j.Logger
-
-case class DataSourceParams(filepath: String) extends Params // CHANGED
-
-class DataSource(val dsp: DataSourceParams)
- extends PDataSource[TrainingData,
- EmptyEvaluationInfo, Query, EmptyActualResult] {
-
- @transient lazy val logger = Logger[this.type]
-
- override
- def readTraining(sc: SparkContext): TrainingData = {
- // CHANGED
- val data = sc.textFile(dsp.filepath)
- val ratings: RDD[Rating] = data.map(_.split("::") match {
- case Array(user, item, rate) =>
- Rating(user, item, rate.toDouble)
- })
- new TrainingData(ratings)
- }
-}
-
-case class Rating(
- user: String,
- item: String,
- rating: Double
-)
-
-class TrainingData(
- val ratings: RDD[Rating]
-) extends Serializable {
- override def toString = {
- s"ratings: [${ratings.count()}] (${ratings.take(2).toList}...)"
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/Engine.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/Engine.scala b/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/Engine.scala
deleted file mode 100644
index 1446ca4..0000000
--- a/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/Engine.scala
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.template.recommendation
-
-import org.apache.predictionio.controller.IEngineFactory
-import org.apache.predictionio.controller.Engine
-
-case class Query(
- user: String,
- num: Int
-)
-
-case class PredictedResult(
- itemScores: Array[ItemScore]
-)
-
-case class ItemScore(
- item: String,
- score: Double
-)
-
-object RecommendationEngine extends IEngineFactory {
- def apply() = {
- new Engine(
- classOf[DataSource],
- classOf[Preparator],
- Map("als" -> classOf[ALSAlgorithm]),
- classOf[Serving])
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/dec9f84c/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/Preparator.scala
----------------------------------------------------------------------
diff --git a/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/Preparator.scala b/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/Preparator.scala
deleted file mode 100644
index 0bab35b..0000000
--- a/examples/experimental/scala-parallel-recommendation-custom-datasource/src/main/scala/Preparator.scala
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.template.recommendation
-
-import org.apache.predictionio.controller.PPreparator
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.rdd.RDD
-
-class Preparator
- extends PPreparator[TrainingData, PreparedData] {
-
- def prepare(sc: SparkContext, trainingData: TrainingData): PreparedData = {
- new PreparedData(ratings = trainingData.ratings)
- }
-}
-
-class PreparedData(
- val ratings: RDD[Rating]
-) extends Serializable