基于Spark ALS在线推荐系统 基于als的 推荐

所用技术：

Bootstrap、flat-ui 、 Servlet、Spark1.4.1、Hadoop2.6.0、JDK

说明：本系统不涉及ssh相关内容，只有简单的Servlet和JSP、HTML页面，系统架构相对简单。

系统部署：

1. 拷贝spark-assembly-1.4.1-hadoop2.6.0.jar到WebContent/WEB-INF/lib目录；

（spark-assembly-1.4.1-hadoop2.6.0.jar文件由原生spark-assembly-1.4.1-hadoop2.6.0.jar删除javax/servlet包获得，由于太大，所以就没有上传了）；

2. 拷贝原生spark-assembly-1.4.1-hadoop2.6.0.jar文件到HDFS（目录和代码中一致）；

3. 拷贝WebContent/WEB-INF/lib目录中的Spark141-als.jar到HDFS（目录和代码中保持一致）；

4. 拷贝Hadoop集群（调用所使用的集群，每个人不一样）配置文件yarn-site.xml到HDFS（目录和代码中保持一致）；

5. 修改相关配置文件，由于hadoop相关配置、系统的一些属性需要修改为实际的配置及属性，所以针对这些需要进行修改（后面版本中会对此单独一个配置文件），例如：

或

系统使用数据为movielens上面的数据，下载地址为：http://grouplens.org/datasets/movielens/ ，本测试使用的数据是：

可以根据自己集群的实际情况选择下载数据集的大小。

代码下载地址：https://github.com/fansy1990/movie_recommend ；

系统界面及相关功能实现

1. 系统首页

系统首页如下图所示：

首页直接使用bootstrap的tab界面，分为三栏，分别对应：首页介绍、算法调用和推荐；

2. 初始化后台任务

在启动tomcat的时候，后台会打印相关日志：

1. 
   
2. 信息: Starting Servlet Engine: Apache Tomcat/7.0.52
3. initial begin...
4. 2016-08-23 12:33:28,189 WARN [org.apache.hadoop.util.NativeCodeLoader] - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
5. 2016-08-23 12:33:29,836 INFO [util.Utils] - Movies data size:3883
6. 2016-08-23 12:33:33,638 INFO [util.Utils] - Users data size:6040
7. initial end!
8. 八月 23, 2016 12:33:33 下午 org.apache.coyote.AbstractProtocol start

这里是初始化的相关打印，初始化使用InitServlet，在里面调用了Utils的init方法，init方法主要初始化了movies变量和userWithRatedMovies变量和allMovieIds变量，各个变量表示意思如下：

• movies：所有的电影ID和电影所有相关信息的Map映射；
• userWithRatedMovies：用户ID和当前用户所有评分过的电影ID集合的Map映射；
• allMovieIds：所有电影ID的Set集合；

这里看到初始化的电影有3883个，而用户数有6040个；

3. 建模前台与后台功能实现

建模界面如下所示：

用户输入或选择对应的参数，即可点击“建模”，提交Spark ALS任务到YARN（Hadoop集群），进行算法调用。

建模流程：

1. 用户输入相关算法参数后，点击建模；
2. 后台RunALS Servlet获取提交的算法参数，封装Spark ALS算法，然后提交给YARN；
3. YARN在分配了相关资源后，会返回一个任务ID：applicationID，这时启动一个线程，专门获取该applicationId的任务进度，更新全局allAppStatus变量（Map变量<applicationId,任务状态>），后台返回前台此applicationId；
4. 前台获取到此applicationId后，如果获取的applicationId为Null，那么就会弹出一个模态框提示建模提交任务失败；否则，会弹出一个进度条模态框（此进度条模态框下面会有详细介绍）；
5. 前台启动定时任务，去后台获取全局allAppStatus变量对应applicationId的状态，返回前台，更新进度条模态框对应进度；
6. 一直到任务成功或失败，给出对应的提示；

3.1进度条模态框实现

1. 进度条模态框div定义如下：

1. 
   
2. <div class="modal fade" id="myModal1" tabindex="-1" role="dialog"
3. aria-labelledby="myModalLabel">
4. <div class="modal-dialog" role="document">
5. <div class="modal-content">
6. <div class="progress progress-striped active"
7. style="margin-bottom: 0px; height: 25px; border-radius: 5px;">
8. <div id="progressId" class="progress-bar"
9. style="width: 1%; height: 100%;">0%</div>
10. </div>
11. </div>
12. </div>
13. </div>

全部使用bootstrap的基本样式；

2. 弹出该模态框以及更新进度相关代码：

1. 
   
2. // 弹出窗提示程序正在运行
3. setProgress("progressId", "0%");
5. // 开启进度条模态框
6. openModal("myModal1");
8. // 定时请求任务进度
9. t=setTimeout("queryTaskProgress('"+ret+"')",1000);

相关函数：

1. 
   
2. /**
3. * 设置进度条
4. * @param id
5. * @param value
6. */
7. function setProgress(id,value){
8. $("#"+id).css("width",value);
9. $("#"+id).html(value);
10. }
12. /**
13. * 开启模态框
14. * @param id
15. */
16. function openModal(id){
17. $('#'+id).on('show.bs.modal', function(){
18. var $this = $(this);
19. var $modal_dialog = $this.find('.modal-dialog');
20. // 关键代码，如没将modal设置为 block，则$modala_dialog.height() 为零
21. $this.css('display', 'block');
22. $modal_dialog.css({'margin-top': Math.max(0, ($(window).height() - $modal_dialog.height()) / 2) });
23. });
24. $('#'+id).modal({backdrop: 'static', keyboard: false});
25. }

定时函数，查看进度：

1. 
   
2. /**
3. * 请求任务进度
4. */
5. function queryTaskProgress(appId){
6. // ajax 发送请求获取任务运行状态，如果返回运行失败或成功则关闭弹框
7. $.ajax({
8. type : "POST",
9. url : "Monitor",
10. //dataType : "json",
11. async:false,// 同步执行
12. data:{APPID:appId},
13. success : function(data) {
14. //console.info("success:"+data);
15. if(data.indexOf("%")==-1){// 不包含 ，任务运行完成（失败或成功）
16. clearTimeout(t);// 关闭计时器
17. // 关闭弹窗进度条
18. $('#myModal1').modal("hide");
19. // 开启提示条模态框
21. $('#tipId').html(data=="FINISHED"?"模型训练完成！":
22. (data=="FAILED"?"调用建模失败!":"模型训练被杀死！"));
24. openModal("myModal2");
25. console.info("closed!");
26. return ;
27. }
29. setProgress("progressId", data);
30. // 进度查询每次间隔1500ms
31. t=setTimeout("queryTaskProgress('"+appId+"')",1500);
32. },
33. error: function(data){
34. console.info("error"+data);
36. }
37. });
38. }

3.2进度条模态框效果

3.2 Eclipse提交Spark任务到YARN后台实现

提交任务参考了部分Spark源码实现，下面是代码：

1. 封装Spark ALS算法程序，准备提交任务到YARN；

1. 
   
2. String[] runArgs=new String[]{
3. "--name","ALS Model Train ",
4. "--class","als.ALSModelTrainer",
5. "--driver-memory","512m",
6. "--num-executors", "2",
7. "--executor-memory", "512m",
8. "--jar","hdfs://master:8020/user/root/Spark141-als.jar",//
9. "--files","hdfs://master:8020/user/root/yarn-site.xml",
10. "--arg",input,
11. "--arg",output,
12. "--arg",train_percent,
13. "--arg",ranks,
14. "--arg",lambda,
15. "--arg",iteration
16. };
17. FileSystem.get(Utils.getConf()).delete(new Path(output), true);
18. return Utils.runSpark(runArgs);

（注意：1. 这里的部分参数应该是需要隔离到配置文件里面的，比如--class 或--driver-memory的值等；2. 本来在allAppStatus中设置的是一个全局变量，所以我本意是可以多用户提交任务，进而监控也是分开的，但是这里会有个问题，就是模型的输出目录，这个应该是需要和用户挂钩，同时在建模的时候，每个用户的推荐也需要采用各自对应的模型，但是目前来说，这个功能有点复杂，暂时就考虑一个用户，一个模型；）

2. 提交Spark任务到YARN，同时开启对应监控，更新任务状态

1. 
   
2. /**
3. * 调用Spark 加入监控模块
4. *
5. * @param args
6. * @return Application ID字符串
7. */
8. public static String runSpark(String[] args) {
9. try {
10. System.setProperty("SPARK_YARN_MODE", "true");
11. SparkConf sparkConf = new SparkConf();
12. sparkConf.set("spark.yarn.jar", "hdfs://master:8020/user/root/spark-assembly-1.4.1-hadoop2.6.0.jar");
13. sparkConf.set("spark.yarn.scheduler.heartbeat.interval-ms", "1000");
15. ClientArguments cArgs = new ClientArguments(args, sparkConf);
17. Client client = new Client(cArgs, getConf(), sparkConf);
18. // client.run(); // 去掉此种调用方式，改为有监控的调用方式
20. /**
21. * 调用Spark ，含有监控
22. */
23. ApplicationId appId = null;
24. try{
25. appId = client.submitApplication();
26. }catch(Throwable e){
27. e.printStackTrace();
28. // 返回null
29. return null;
30. }
31. // 开启监控线程
32. updateAppStatus(appId.toString(),"2%" );// 提交任务完成，返回2%
33. log.info(allAppStatus.toString());
34. new Thread(new MonitorThread(appId,client)).start();
35. return appId.toString();
36. } catch (Exception e) {
37. e.printStackTrace();
38. return null;
39. }
40. }

之前直接使用Client的run方法，提交任务，但是这样就获取不到applicationId，如下图

所以就去掉这种方式，参考Client中的run方法的具体实现，编写对应代码来进行任务提交；（需要注意这种提交方式，当任务失败或完成后，需要删除相关临时文件）；

后台监控：

相关代码，在更新任务状态时进行：

1. 
   
2. // 完成/ 失败/杀死
3. if (state == YarnApplicationState.FINISHED || state == YarnApplicationState.FAILED
4. || state == YarnApplicationState.KILLED) {
5. Utils.cleanupStagingDir(appId);
6. // return (state, report.getFinalApplicationStatus);
7. // 更新 app状态
8. log.info("Thread:"+Thread.currentThread().getName()+
9. appId.toString()+"完成，任务状态是："+state.name());
10. Utils.updateAppStatus(appId.toString(), state.name());
11. return;
12. }

该代码在MonitorThread中；（但是，需要注意的是，如果Spark任务正在运行，那么这时关闭Tomcat，就会导致相关临时文件删除不了，为什么？请大家自己思考）

4. 推荐前台与后台功能实现

4.1 推荐页面前台

前台界面如下：

前台有两个功能，一个功能是输入用户ID，查询出当前用户ID评分过的电影信息；一个功能是根据用户ID和推荐个数，对用户进行电影推荐；

查询功能结果：

这里需要注意，评分全部为零，这个是因为在userRatedMovieIds这个变量中存储的只是用户的评分过的电影ID，并没有附加评分，所以可以在这个地方进行修改，以显示正确的电影评分（同时，这里的查询，也可以把所有信息存储在HBase中，进行查询）；

推荐功能结果：

推荐功能展示的结果，是按照推荐分降序排列的；

不管是查询还是推荐，前台直接使用一个div来接收这些信息：

<div class="col-md-10 col-md-offset-1" id="movieResultId">

接着使用AJax获取后台对应的数据进行拼接，在赋值给div：

1. 
   
2. // 绑定推荐button
3. $("#recommendId").click(function(){
4. var userId = $('#userId').val();
5. var recommendNum = $('#recommendNumId').val();
6. var ret =null;
7. $.ajax({
8. type : "POST",
9. url : "Recommend",
10. async:false,// 同步执行
11. data : {userId:userId,flag:"recommend",recommendNum:recommendNum},
12. //dataType : "json",
13. success : function(data) {// data 返回appId
14. ret = data;
15. },
16. error: function(data){
17. console.info("error"+data);
18. ret = data=="null"?"null":data;
19. }
20. });
23. var showResultHtml = '<br>'+
24. '<p>数据如下：</p>'+
25. '<div class="table-responsive">' +
26. '<table class="table table-striped">' +
27. '<thead>'+
28. '<tr>'+
29. '<th>MovieId</th>'+
30. '<th>电影名</th>'+
31. '<th>标签</th>'+
32. '<th>推荐分</th>'+
33. '</tr>'+
34. '</thead>'+
35. '<tbody>'+
36. ret +
37. '</tbody>'+
38. '</table>'+
39. '</div>';
40. $('#movieResultId').html(showResultHtml);
41. });

4.2 推荐页面后台

推荐页面的查询，只是简单的Map的数据获取而已；重点是推荐功能。

推荐功能最开始我想的是直接保存Spark ALS的模型，然后调用Spark ALS模型的predict（user,product），即可直接得到用户的推荐分，但是这样是不行的：

参考：http://stackoverflow.com/questions/34288435/using-java-for-running-mllib-model-with-streaming ；Spark里面的模型有些是本地的有些是分布式的，如果是分布式的，那么是不能执行类似predict操作的，而Spark ALS的模型MatrixFactorizationModel 是分布式的，所以不能够直接执行predict操作。这里同样是参考Spark的源码，来进行的。

在建模完成后，把Spark ALS模型的两个参数userFeatures、productFeatures分别存入HDFS，然后在模型推荐的时候把其加载进内存，使用userFeatures和productFeatures两个变量即可完成推荐：

1. 
   
2. /**
3. * 预测 如果没有初始化，则进行初始化
4. *
5. * @param uid
6. * @param recNum
7. * @return
8. * @throws NoSuchMethodException
9. * @throws InvocationTargetException
10. * @throws InstantiationException
11. * @throws IllegalAccessException
12. */
13. public static List<Movie> predict(int uid,int recNum) throws IllegalAccessException, InstantiationException, InvocationTargetException, NoSuchMethodException {
14. if (userFeatures.size() <= 0 || productFeatures.size() <= 0) {
15. try {
16. userFeatures = getModelFeatures(userFeaturePath);
17. productFeatures = getModelFeatures(productFeaturePath);
18. } catch (IOException e) {
19. return null;
20. }
21. if (userFeatures.size() <= 0 || productFeatures.size() <= 0) {
22. System.err.println("模型加载失败!");
23. return null;
24. }
25. }
27. // 使用模型进行预测
28. // 1. 找到uid没有评价过的movieIds
29. Set<Integer> candidates = Sets.difference((Set<Integer>) allMovieIds, userWithRatedMovies.get(uid));
31. // 2. 构造推荐排序堆栈
32. FixSizePriorityQueue<Movie> recommend = new FixSizePriorityQueue<Movie>(recNum);
33. Movie movie = null;
34. double[] pFeature = null;
35. double[] uFeature = userFeatures.get(uid);
36. double score = 0.0;
37. BLAS blas = BLAS.getInstance();
38. for (int candidate : candidates) {
39. movie = movies.get(candidate).deepCopy();
40. pFeature = productFeatures.get(candidate);
41. if (pFeature == null)
42. continue;
43. score = blas.ddot(pFeature.length, uFeature, 1, pFeature, 1);
44. movie.setRated((float) score);
45. recommend.add(movie);
46. }
48. return recommend.sortedList();
49. }

中间的score= blas.ddot就是计算推荐分的；

总结

1. 基本完成相关推荐系统功能；

2. 相关参数需要额外添加配置文件，而不是直接硬编码到代码中；

3. 推荐只能针对已经存在的用，不能进行匿名推荐（同时使用SPark ALS模型推荐的结果基本一样，这个是Spark的bug？还是调用哪里有问题？）；

4. 添加多用户调用支持；

5. 查询用户评分过的功能完善（对应评分获取）；