1.觸發(fā)條件 接受到用戶評(píng)分大于3.5時(shí)，寫入日志 system.log
//獲取用戶的評(píng)分?jǐn)?shù)據(jù)
//電影評(píng)星
@RequestMapping(value = "/getStar", method = RequestMethod.POST)
@ResponseBody
public String getStar(HttpServletRequest request) throws Exception {
HttpSession session = request.getSession();
int movie_id = Integer.parseInt(request.getParameter("movie_id"));
Double star = Double.parseDouble(request.getParameter("star"));
String username = (String) session.getAttribute("username");
//緩存到數(shù)據(jù)庫（mysql和redis）
int userId = userService.getUserIdByUsername(username);
//獲取當(dāng)前時(shí)間戳
int time = (int)new Date().getTime();
ratingsService.updateRating(new Rating(userId,movie_id,star,time));
//當(dāng)前評(píng)分大于3.5表示用戶對(duì)當(dāng)前評(píng)分的電影感興趣就寫入日志進(jìn)行監(jiān)聽
if (star>=Constant.USERLIKESTAR){
logger.info(Constant.MOVIE_RATING_PREFIX+userId+"|"+movie_id+"|"+star);
}
return "success";
}
日志文件配置
log4j.rootLogger=INFO, file, stdout
# write to stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %5p --- [%50t] %-80c(line:%5L) : %m%n
# write to file
log4j.appender.file=org.apache.log4j.RollingFileAppender
log4j.appender.file.File=D:\\projects\\GraduationProgram\\src\\main\\resources\\logs\\system.log
log4j.appender.file.MaxFileSize=1024KB
log4j.appender.file.MaxBackupIndex=1
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %5p --- [%50t] %-80c(line:%6L) : %m%n
2.Fflume監(jiān)聽日志，F(xiàn)lume配置在本地windows 將監(jiān)聽的日志發(fā)送到CentOS的Kafka中
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -f D:/projects/GraduationProgram/src/main/resources/logs/system.log
a1.sources.r1.fileHeader = true
a1.sources.r1.deserializer.outputCharset=UTF-8
# Describe the sink
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = recommender
a1.sinks.k1.brokerList = 192.168.100.112:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.k1.kafka.producer.compression.type = snappy
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
Spark Streaming處理流程
package org.cuit.recommend.streamingRecommend
import java.io.FileInputStream
import java.util.{Calendar, Date, Properties, TimeZone, Timer}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Dataset, SparkSession}
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.cuit.recommend.offlineRecommend.OfflineRecommend.Ratings
import org.cuit.recommend.offlineStatistics.OfflineStatistics.Ratings
import redis.clients.jedis.Jedis
/**
* @Author
* @Date 2022/3/28 16:39
* @version 1.0
* @注釋
*/
//連接工具類
object ConnectUtil extends Serializable{
val properties = new Properties()
val path = Thread.currentThread().getContextClassLoader.getResource("config.properties").getPath //文件要放到resource文件夾下
properties.load(new FileInputStream(path))
//獲取相關(guān)數(shù)據(jù)表名字
val REDIS_HOST = properties.getProperty("redis_host")
lazy val jedis = new Jedis(REDIS_HOST, 6379)
}
object StreamingRecommend {
/**
* 電影相似推薦矩陣
*
* @param movie_id 電影id
* @param movie_sim_info 相似的電影id+ 相似得分
*/
case class MovieSim(movie_id: Int, movie_sim_info: String)
/**
* 用戶的推薦列表
*
* @param user_id 用戶id
*
* @param user_sim_info 電影id+推薦評(píng)分
*
*/
case class UserSim(user_id: Int, user_sim_info:String)
/**
* 用戶實(shí)時(shí)推薦列表
* @param user_id 用戶id
* @param streaming_recom_info 推薦列表
*/
case class StreamingRecommend(user_id: Int, streaming_recom_info:String)
//用戶評(píng)價(jià)
//1,31,2.5,1260759144
/**
*
* @param user_id 用戶編碼
* @param movie_id imdb編碼
* @param rating 評(píng)級(jí)
* @param rating_timestamp 時(shí)間戳
*/
case class Ratings(user_id: Int, movie_id: Int, rating: Double, rating_timestamp: Int)
case class MysqlConfig(driver:String,url:String,username:String,password:String,tableName:String)
//讀取配置文件
val properties = new Properties()
val path = Thread.currentThread().getContextClassLoader.getResource("config.properties").getPath //文件要放到resource文件夾下
properties.load(new FileInputStream(path))
//獲取相關(guān)數(shù)據(jù)表名字
val MOVIES_TABLE = properties.getProperty("movies_table")
val RATING_TABLE = properties.getProperty("rating_table")
val MOVIE_SIM_TABLE = properties.getProperty("movieSim_table")
val USER_SIM_TABLE = properties.getProperty("userSim_table")
val STREAMING_RECOMMEND_TABLE = properties.getProperty("streaming_Recommend_table")
val MAX_RECENT_RATING_NUM = properties.getProperty("max_recent_rating_num")
val MAX_MOVIE_SIM_NUM =properties.getProperty("max_movie_sim_num")
val KAFKA_TOPIC =properties.getProperty("Kafka_topic")
val MYSQL_URI = properties.getProperty("musql_url")
val KAFKA_BOOTSTRAP_SERVERS = properties.getProperty("kafka_bootstrap_servers")
val KAFKA_GROUP_ID = properties.getProperty("kafka_group_id")
val prop = new java.util.Properties
val USER = properties.getProperty("jdbc_username")
val PASSWORD = properties.getProperty("jdbc_password")
val LOG_PARAMETER = properties.getProperty("log_parameter")
val DRIVER = properties.getProperty("driver")
val mysqlConfig: MysqlConfig = MysqlConfig(DRIVER, MYSQL_URI, USER, PASSWORD,STREAMING_RECOMMEND_TABLE)
def streamingRecommend():Unit={
//設(shè)置連接對(duì)象
//拼接連接的url
var url = s"${MYSQL_URI}?user=${USER}&password=${PASSWORD}"
prop.setProperty("user", USER)
prop.setProperty("password", PASSWORD)
//創(chuàng)建SparkConf
val conf: SparkConf = new SparkConf().setAppName("StreamingRecommend").setMaster("local[*]")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
//創(chuàng)建SparkSession
val sc = new SparkContext(conf)
val sparkSession = SparkSession.builder().config(conf).getOrCreate()
//StreamingContext 2:批處理時(shí)間
val ssc: StreamingContext = new StreamingContext(sc, Seconds(2))
//隱式轉(zhuǎn)換
import sparkSession.implicits._
//獲取數(shù)據(jù) 由于數(shù)據(jù)規(guī)模龐大，性能要求，將其廣播出去
//得到一個(gè)評(píng)分map 方便后面直接根據(jù)user_id key 獲取已經(jīng)評(píng)分的movie_id s
val ratingsMap: Map[Int, Array[Int]] = sparkSession.read
//url /127.0.0.1:3306/test?user=root&password=root
.option("url", url)
//表名
.option("dbtable", RATING_TABLE)
.format("jdbc")
.load()
//轉(zhuǎn)換為ratings類
.map{
row=>(row.getAs[Int]("user_id"),row.getAs[Int]("movie_id"))
}.collect().groupBy(_._1).map{
//user_id (movie_id,Movie_id........)
row=>(row._1,(row._2.map(_._2)))
}
//獲取數(shù)據(jù)
val movieSimMap: collection.Map[Int, Map[Int, Double]] = sparkSession.read
//url /127.0.0.1:3306/test?user=root&password=root
.option("url", url)
//表名
.option("dbtable", MOVIE_SIM_TABLE)
.format("jdbc")
.load()
//轉(zhuǎn)換為ratings類
.as[MovieSim]
.rdd
//格式轉(zhuǎn)換
.map{ movie =>{
//類型變換 類型轉(zhuǎn)換 + map(map()) movie_Id為key 方便后面根據(jù)兩個(gè)movie_id查詢
//movieRec: (movie_id,movie_id(String):score(String)| ....) ->
// (map(movie_id(key),List(movie_id(Int),score(Double)) ... ))
// -> (map(movie_id(key),List(map(movie_id(Int)(key),score(Double))) ... ))
//(movie_id,(array)) split:做切分得到數(shù)組之后對(duì)每個(gè)元素進(jìn)行處理 movie_sim_info：|5319 : 0.742| 27803:0.661| 6370:0.648| 40583:0.643|
(movie.movie_id,movie.movie_sim_info.split("\\|").map{
x=>{ //|5319 : 0.742
val info = x.split(":")
(info(0).toInt,info(1).toDouble)
}
}.toMap)
}
}.collectAsMap()
val movieSimBroadCast: Broadcast[collection.Map[Int, Map[Int, Double]]] = sc.broadcast(movieSimMap)
//進(jìn)行流式處理
//定義Kafka參數(shù)
val kafkaParams = Map[String, Object](
//連接參數(shù) 109.168.100。112:9092
"bootstrap.servers" -> KAFKA_BOOTSTRAP_SERVERS,
//序列化
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> KAFKA_GROUP_ID,
"auto.offset.reset" -> "latest",
//自動(dòng)提交
"enable.auto.commit" -> (false: java.lang.Boolean)
)
//通過Kafka創(chuàng)建一個(gè)KafkaStream
val kafkaStream = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe[String, String](Array(KAFKA_TOPIC), kafkaParams)
)
//日志數(shù)據(jù) user_id|movie_id|rating|rating——timestamp 轉(zhuǎn)換為評(píng)分流數(shù)據(jù)
val userDataStream: DStream[(Int, Int, Double, Int)] = kafkaStream.map {
//獲取評(píng)分值
data => {
val user_data = data.value().split("RATING_PREFIX:")(1).split("\\|")
// user_id INT (6) not null,
// movie_id INT not null,
// rating FLOAT,
// rating_timestamp INT
(user_data(0).toInt, user_data(1).toInt, user_data(2).toDouble, new Date().getTime.toInt)
}
}
/**
*實(shí)時(shí)計(jì)算，根據(jù)得到的信息進(jìn)行實(shí)時(shí)計(jì)算
*/
//對(duì)窗口獲取到的一組用戶數(shù)據(jù)進(jìn)行遍歷
userDataStream.foreachRDD {
//對(duì)每一條數(shù)據(jù)進(jìn)行模式匹配
userData =>userData.foreach{
case (user_id:Int, movie_id:Int, rating:Double, rating_timestamp:Int) => {
println(s"data --::--- user_id = ${user_id},movie_id = ${movie_id},rating=${rating},rating_timestamp=${rating_timestamp}")
//數(shù)據(jù)更新到redis和mysql
/**
* 流程
* 1.獲取用戶的k次評(píng)分 <-- redis movie_id,rating
* 2.根據(jù)當(dāng)前用戶評(píng)分的電影 取得 與該電影 最相似的電影作為推薦列表 movie_id......
* 3.計(jì)算電影的優(yōu)先推薦等級(jí)
* 4.根據(jù)優(yōu)先級(jí)獲得推薦列表 movie_id,score
* 5.緩存到業(yè)務(wù)數(shù)據(jù)庫mysql
*/
//1.獲取用戶的k次評(píng)分 <-- redis
val recentRating: Array[(Int, Double)] = RecommendService.getUserRecentRatingFromRedis(MAX_RECENT_RATING_NUM.toInt, user_id, ConnectUtil.jedis)
//2根據(jù)當(dāng)前用戶評(píng)分的電影 取得 與該電影 最相似的電影作為推薦列表 movie_id......
//2.1篩選出用戶已經(jīng)評(píng)價(jià)過的電影
val userAlreadyRating = ratingsMap(user_id)
// val userAlreadyRating: Array[Int] = ratingsRdd.filter {
// rating => rating.user_id == user_id
// //return rdd[Ratings]
// }.map(_.movie_id).collect() //將篩選后的返回一個(gè)mivie_id列表
//2.2 根據(jù)當(dāng)前用戶評(píng)分的電影 取得 與該電影 最相似的電影作為推薦列表
val movieRecommendList: Array[Int] = RecommendService.getMovieSimListFromMysql(movie_id, userAlreadyRating, MAX_MOVIE_SIM_NUM.toInt, movieSimBroadCast.value)
//3,4.計(jì)算電影的優(yōu)先推薦等級(jí)并得到列表 recentRating最近的評(píng)分 movieRecommendList：待推薦的電影列表 movieSimBroadCast：獲取近期推薦電影的評(píng)分
val streamingRecs: Array[(Int, Double)] = RecommendService.calculateRecommendPriority(LOG_PARAMETER.toInt, recentRating, movieRecommendList, movieSimBroadCast.value)
//4.緩存到業(yè)務(wù)數(shù)據(jù)庫mysql
// RecommendService.storeResToMysql(streamingRecs, user_id)(mysqlConfig)
RecommendService.storeResToRedis(streamingRecs, user_id,ConnectUtil.jedis)
}
}
}
//準(zhǔn)備接受數(shù)據(jù)流
ssc.start()
//等待時(shí)間
ssc.awaitTermination()
}
def main(args: Array[String]): Unit = {
streamingRecommend()
}
}
處理函數(shù)
package org.cuit.recommend.streamingRecommend
import java.sql.{Connection, DriverManager, ResultSet}
import com.mysql.jdbc.PreparedStatement
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils
import org.cuit.recommend.offlineStatistics.OfflineStatistics.Ratings
import org.cuit.recommend.streamingRecommend.StreamingRecommend.{MAX_MOVIE_SIM_NUM, MAX_RECENT_RATING_NUM, MysqlConfig}
import redis.clients.jedis.Jedis
import scala.collection.mutable.ArrayBuffer
/**
* @Author 高拴梁
* @Date 2022/4/25 16:13
* @version 1.0
* @注釋 實(shí)時(shí)推薦業(yè)務(wù)的具體實(shí)現(xiàn)
* * 1.獲取用戶的k次評(píng)分 <-- redis movie_id,rating
* * 2.根據(jù)當(dāng)前用戶評(píng)分的電影 取得 與該電影 最相似的電影作為推薦列表 movie_id......
* * 3.計(jì)算電影的優(yōu)先推薦等級(jí)
* * 4.根據(jù)優(yōu)先級(jí)獲得推薦列表 movie_id,score
* * 5.緩存到業(yè)務(wù)數(shù)據(jù)庫mysql
*/
object RecommendService {
/**
*
* @param recentRatingNum 實(shí)時(shí)推薦選擇的最近k次評(píng)分
* @param user_id 用戶id
* @param jedis redis連接客戶端
* @return 最近評(píng)分結(jié)果(movie_id,rating)
*/
def getUserRecentRatingFromRedis(recentRatingNum:Int,user_id:Int,jedis:Jedis):Array[(Int,Double)]={
jedis.auth("123456")
//jedis.lrange 返回Java的Array類型，需要轉(zhuǎn)換才能使用Scala的map
import scala.collection.JavaConversions._
//redis獲取數(shù)據(jù) key:user_id value: movie_id:rating "1900:4.5"....
// 返回列表中指定區(qū)間內(nèi)的元素，區(qū)間以偏移量 START 和 END 指定
jedis.lrange("user_id:"+user_id,0,recentRatingNum).map{
x=> {
val userData = x.split(":")
(userData(0).toInt,userData(0).toDouble)
}
}.toArray
}
/**
* 從電影相似度矩陣獲取最相似的n個(gè)電影 (不包括用戶已經(jīng)瀏覽評(píng)分過的電影)
* @param movie_id 當(dāng)前評(píng)分的電影id
* @param alreadyRating 用戶已經(jīng)評(píng)價(jià)過的電影
* @param movieSimNum 獲取的電影數(shù)量
* @param movieSim 電影相似度矩陣
* @return 最相似的電影id Array(1,2,3,4.....)
*/
def getMovieSimListFromMysql(movie_id:Int
,alreadyRating:Array[Int]
,movieSimNum:Int
,movieSim:scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]):Array[Int]={
/**
* 考慮到推薦結(jié)果不應(yīng)該包含用戶評(píng)分過的電影
* 流程
* 1.根據(jù)movie_id從movieSimBroadCast獲取最相似的n個(gè)電影
* 2.過濾掉movie_id相同的電影
*/
//movieSim Map[Int(movie_id), Map[Int(movie_id), Double(sim_score)]] =>Array(Int(movie_id), Double(sim_score))
val simMovies = movieSim(movie_id).toArray
//2.過濾掉看過的電影 排序取
simMovies.filter(movie=> !alreadyRating.contains(movie._1))
.sortWith(_._2>_._2)
.take(movieSimNum)
//獲得電影列表
.map(_._1)
}
/**
* 計(jì)算實(shí)時(shí)推薦的電影的優(yōu)先級(jí)并返回優(yōu)先級(jí)結(jié)果
* @param logParameter 求log的底數(shù)
* @param recentRating 最近評(píng)分結(jié)果(movie_id,rating)
* @param movieRecommendList 最相似的電影id Array(1,2,3,4.....)
* @param movieSimBroadCast 電影相似度矩陣
* @return 實(shí)施推薦結(jié)果列表 Array((movie_id,score)))
*/
def calculateRecommendPriority(logParameter:Int
,recentRating: Array[(Int, Double)]
,movieRecommendList: Array[Int]
,movieSimBroadCast:scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]):Array[(Int,Double)]={
//計(jì)算每個(gè)推薦電影（推薦電影來自于movieRecommendList 即電影相似度矩陣的topN）的推薦得分 Int, Double movie_id,score
val movieScore = new ArrayBuffer[(Int, Double)]()
//推薦優(yōu)先級(jí) incount Int ,Int:電影id,用戶最近k次評(píng)分中大于某個(gè)值的個(gè)數(shù) 個(gè)數(shù)越多說明該電影越值得推薦
val inCountMap = scala.collection.mutable.HashMap[Int, Int]()
//不推薦優(yōu)先級(jí) decount 同理
val deCountMap = scala.collection.mutable.HashMap[Int, Int]()
//遍歷待推薦電影和用戶近期的k次評(píng)分電影 得到基礎(chǔ)得分movieScore中的值
//((1,0.5),(1,0.6),(1,0.7)...) 1:待推薦的電影id 0.5:該待推薦電影與近期評(píng)分?jǐn)?shù)據(jù)的推薦值
for (movieRecommend<-movieRecommendList;recentMovie<-recentRating){
//計(jì)算兩者的相似度，給定兩個(gè)mid從相似度矩陣中得到對(duì)應(yīng)的值
val score = getMovieSimScoreFromMovieSimRecs(movieRecommend, recentMovie._1,movieSimBroadCast)
if (score>0.5){
//sim(q,r) * Rating r (movieRecommend,score * recentMovie._2)作為一個(gè)整體元組放入ArrayBuffer
movieScore += ((movieRecommend,score * recentMovie._2))
if (recentMovie._2>3){
//避免因未初始化而報(bào)錯(cuò)
inCountMap(movieRecommend) = inCountMap.getOrElse(movieRecommend,0)+1
}else deCountMap(movieRecommend) = deCountMap.getOrElse(movieRecommend,0)+1
}
}
//做聚合求最后的推薦值 movie_id聚合 ((1,0.5),(1,0.6),(1,0.7)...) 1:待推薦的電影id 0.5:該待推薦電影與近期評(píng)分?jǐn)?shù)據(jù)的推薦值
movieScore.groupBy(_._1).map{
/**
* 推薦值 = sim(q,r) * Rating r 做聚合(按照movie_id)求均值 + lg max(incount,1) - lg max(decount,1)
*/
//模式匹配 返回(movie_id,推薦值) scoreMapList:groupBy后返回的 數(shù)據(jù)類型 Map(movie_id,ArrayBuffer(movie_id,score))
case (movie_id: Int,scoreArray) => (movie_id,scoreArray.map(_._2).sum / scoreArray.length
+ log(inCountMap.getOrElse(movie_id,1))-log(deCountMap.getOrElse(movie_id,1)))
}.toArray.sortWith(_._2>_._2)
}
/**
* 插入結(jié)果數(shù)據(jù)到mysql
* @param streamingRecs 該用戶推薦的電影(movie_id,score),(),()
* @param user_id 用戶id
* @param mysqlConfig mysql連接配置
*/
def storeResToMysql(streamingRecs:Array[(Int,Double)],user_id:Int)(implicit mysqlConfig: MysqlConfig):Unit={
val driver = mysqlConfig.driver
val url = mysqlConfig.url
val username = mysqlConfig.username
val password = mysqlConfig.password
var connection:Connection = null
//結(jié)果處理
val streamingRecommendinfo: String = streamingRecs.map {
case (movie_id: Int, score: Double) => s"${movie_id}:${score.formatted("%.3f")}|"
}.mkString("")
try {
Class.forName(driver)
connection = DriverManager.getConnection(url, username, password)
val statement = connection.createStatement()
//判斷未更新的原始數(shù)據(jù)是否存在
val resultSet = statement.executeQuery( s"select * from ${mysqlConfig.tableName}")
while ( resultSet.next() ) {
val deleteSql = s"delete from ${mysqlConfig.tableName} where user_id = ${user_id}"
connection.prepareStatement(deleteSql).executeUpdate()
}
//如果沒有就插入數(shù)據(jù)
val insertSQL="insert into "+mysqlConfig.tableName+" values("+user_id+",\""+streamingRecommendinfo+"\")"
connection.prepareStatement(insertSQL).executeUpdate()
} catch {
case e => e.printStackTrace
//case _: Throwable => println("ERROR")
}
connection.close()
}
/**
* 推薦結(jié)果放入Redis
* @param streamingRecs 實(shí)時(shí)推薦矩陣
* @param user_id 用戶id
* @param jedis jedis配置連接
*/
def storeResToRedis(streamingRecs:Array[(Int,Double)],user_id:Int,jedis: Jedis):Unit={
//結(jié)果處理
val streamingRecommendinfo: String = streamingRecs.map {
case (movie_id: Int, score: Double) => s"${movie_id}:${score.formatted("%.3f")}|"
}.mkString("")
jedis.auth("123456")
jedis.hset("streaming",user_id.toString,streamingRecommendinfo)
}
/**
* 根據(jù)movieA和B的電影id來從movieSimBroadCast中取得兩個(gè)電影的相似度 ，找不到取默認(rèn)值
* @param movieA 電影A的id
* @param movieB 電影B的id
* @param movieSimBroadCast 電影相似度矩陣 Map[Int(MovieA_id), Map[Int(movieB_id), Double(相似度)]
* @return 相似度
*/
def getMovieSimScoreFromMovieSimRecs(movieA:Int,movieB:Int,movieSimBroadCast:scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]):Double={
movieSimBroadCast.get(movieA) match {
case Some(movieSimmap) => movieSimmap.get(movieB) match {
case Some(score) => score
case None => 0.0
}
case None => 0.0
}
}
/**
*
* @param m 代求數(shù)
* @param logParameter 底數(shù)
* @return log結(jié)果
*/
def log(m:Int,logParameter:Int = 10):Double={
//對(duì)數(shù)的換底公式
math.log(m) / math.log(logParameter)
}
}
