Initial commit

Author: muencseb

pom.xml

@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>ch.mue.seb</groupId>
+    <artifactId>javarank</artifactId>
+    <version>1.0-SNAPSHOT</version>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>8</source>
+                    <target>8</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_2.10</artifactId>
+            <version>1.6.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-sql_2.10</artifactId>
+            <version>1.6.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-mllib_2.10</artifactId>
+            <version>1.6.1</version>
+        </dependency>
+    </dependencies>
+
+
+</project>

src/main/java/recommendation/data/InputRating.java

@@ -0,0 +1,37 @@
+package recommendation.data;
+
+public class InputRating {
+    private Integer userId;
+    private Integer productId;
+    private Integer rating;
+
+    public InputRating(Integer userId, Integer productId, Integer rating) {
+        this.userId = userId;
+        this.productId = productId;
+        this.rating = rating;
+    }
+
+    public Integer getUserId() {
+        return userId;
+    }
+
+    public void setUserId(Integer userId) {
+        this.userId = userId;
+    }
+
+    public Integer getProductId() {
+        return productId;
+    }
+
+    public void setProductId(Integer productId) {
+        this.productId = productId;
+    }
+
+    public Integer getRating() {
+        return rating;
+    }
+
+    public void setRating(Integer rating) {
+        this.rating = rating;
+    }
+}

src/main/java/recommendation/data/RDDHelper.java

@@ -0,0 +1,25 @@
+package recommendation.data;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+
+import java.util.Collection;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public class RDDHelper {
+
+    private static JavaSparkContext jsc;
+
+    public RDDHelper(JavaSparkContext javaSparkContext) {
+        jsc = javaSparkContext;
+    }
+
+    public <T> JavaRDD<T> getRddFromCollection(Collection<T> collection) {
+        return jsc.parallelize(collection.stream().parallel().collect(Collectors.toList())).cache();
+    }
+
+    public <T> JavaRDD<T> getRddFromCollection(Collection<?> collection, Function<? super Object, ? extends T> mapper) {
+        return getRddFromCollection(collection.stream().parallel().map(mapper).collect(Collectors.toList()));
+    }
+}

src/main/java/recommendation/exceptions/ModelNotReadyException.java

@@ -0,0 +1,4 @@
+package recommendation.exceptions;
+
+public class ModelNotReadyException extends Exception {
+}

src/main/java/recommendation/model/RecommendationMlModel.java

@@ -0,0 +1,65 @@
+package recommendation.model;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.mllib.recommendation.ALS;
+import org.apache.spark.mllib.recommendation.MatrixFactorizationModel;
+import org.apache.spark.mllib.recommendation.Rating;
+import recommendation.data.InputRating;
+import recommendation.data.RDDHelper;
+import recommendation.exceptions.ModelNotReadyException;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+public class RecommendationMlModel {
+
+    public static final String SPARK_APP_NAME = "Recommendation Engine";
+    public static final String SPARK_MASTER = "local";
+
+    private ALS als = new ALS();
+    private MatrixFactorizationModel model;
+
+    private RDDHelper rddHelper = new RDDHelper(new JavaSparkContext(SPARK_MASTER, SPARK_APP_NAME));
+
+    private ReentrantReadWriteLock mutex = new ReentrantReadWriteLock();
+
+    private boolean modelIsReady = false;
+
+    public boolean isModelReady() {
+        return modelIsReady;
+    }
+
+    public Double getInterestPrediction(Integer userId, Integer eventId) throws ModelNotReadyException {
+        if (!modelIsReady)
+            throw new ModelNotReadyException();
+        mutex.readLock().lock();
+        Double prediction = model.predict(userId, eventId);
+        mutex.readLock().unlock();
+        return prediction;
+    }
+
+    public void createModel(Collection<InputRating> ratings) {
+        trainModel(ratings);
+    }
+
+    private void trainModel(Collection<InputRating> ratings) {
+        JavaRDD<Rating> ratingRDD = rddHelper.getRddFromCollection(createSparkRating(ratings)).cache();
+        if (ratingRDD.isEmpty())
+            return;
+        mutex.writeLock().lock();
+        model = als.setRank(10).setIterations(10).run(ratingRDD);
+        mutex.writeLock().unlock();
+        modelIsReady = true;
+    }
+
+
+    private List<Rating> createSparkRating(Collection<InputRating> inputRatings) {
+        return inputRatings
+                .stream()
+                .map(ir -> new Rating(ir.getUserId(), ir.getProductId(), ir.getProductId()))
+                .collect(Collectors.toList());
+    }
+}

src/main/java/recommendation/service/RecommendationService.java

@@ -0,0 +1,26 @@
+package recommendation.service;
+
+import recommendation.data.InputRating;
+import recommendation.exceptions.ModelNotReadyException;
+import recommendation.model.RecommendationMlModel;
+
+import java.util.Collection;
+
+public class RecommendationService {
+
+    RecommendationMlModel recommendationMlModel = new RecommendationMlModel();
+
+    public RecommendationService(Collection<InputRating> ratings) {
+        recommendationMlModel.createModel(ratings);
+    }
+
+    public boolean isModelReady() {
+        return recommendationMlModel.isModelReady();
+    }
+
+    public Double getPrediction(Integer userId, Integer productId) throws ModelNotReadyException {
+        return recommendationMlModel.getInterestPrediction(userId, productId);
+    }
+
+
+}