大数据技术之电商推荐系统(5) | 离线推荐服务

川长思鸟来 2022-12-21 04:49 297阅读 0赞

### 目录 ###

*  一.离线统计服务
 *   *  统计服务主体框架
     *  需求
 *  二.扩展写到Mysql里

# 一.离线统计服务 #

*  离线推荐服务是综合用户所有的历史数据，利用设定的离线统计算法和离线推荐算法周  
    期性的进行结果统计与保存，计算的结果在一定时间周期内是固定不变的，变更的频率取决于算法调度的频率。
 *  离线推荐服务主要计算一些可以预先进行统计和计算的指标，为实时计算和前端业务相应提供数据支撑。
 *  离线推荐服务主要分为统计推荐、基于隐语义模型的协同过滤推荐以及基于内容和基于Item-CF的相似推荐。我们这一章主要介绍前两部分，基于内容和Item-CF的推荐在整体结构和实现上是类似的。

## 统计服务主体框架 ##

在recommender下新建子项目StatisticsRecommender，pom.xml文件中只需引入spark、scala和mongodb的相关依赖：

<dependencies>
            
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-core_2.11</artifactId>
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-sql_2.11</artifactId>
            </dependency>
            
            <dependency>
                <groupId>org.scala-lang</groupId>
                <artifactId>scala-library</artifactId>
            </dependency>
            
            
            <dependency>
                <groupId>org.mongodb</groupId>
                <artifactId>casbah-core_2.11</artifactId>
                <version>${ casbah.version}</version>
            </dependency>
            
            <dependency>
                <groupId>org.mongodb.spark</groupId>
                <artifactId>mongo-spark-connector_2.11</artifactId>
                <version>${ mongodb-spark.version}</version>
            </dependency>
        </dependencies>

在resources文件夹下引入log4j.properties，然后在src/main/scala下新建scala 单例对象com.donglin.statistics.StatisticsRecommender。  
同样，我们应该先建好样例类，在main()方法中定义配置、创建SparkSession并加载数据，最后关闭spark。代码如下：  
src/main/scala/com.donglin.statistics/StatisticsRecommender.scala

## 需求 ##

*  历史热门商品统计  
    根据所有历史评分数据，计算历史评分次数最多的商品。  
    实现思路：  
    通过Spark SQL读取评分数据集，统计所有评分中评分数最多的商品，然后按照从大到小排序
 *  最近热门商品统计  
    根据评分，按月为单位计算最近时间的月份里面评分数最多的商品集合。  
    实现思路：  
    通过Spark SQL读取评分数据集，通过UDF函数将评分的数据时间修改为月，然后统计每月商品的评分数
 *  商品平均得分统计  
    根据历史数据中所有用户对商品的评分，周期性的计算每个商品的平均得分。  
    实现思路：  
    通过Spark SQL读取保存在MongDB中的Rating数据集

import java.sql.Date
    import java.text.SimpleDateFormat
    
    import org.apache.spark.SparkConf
    import org.apache.spark.sql.{ DataFrame, SparkSession}
    
    case class Rating(userId:Int,productId:Int,socre:Double,timestamp:Int)
    case class MongoConfig(uri:String,db:String)
    
    object StatisticRecommender { 
      //定义mongodb存储的表名
      val MONGODB_RATING_COLLECTION = "Rating"
    
      //统计的表的名称
      val RATE_MORE_PRODUCTS = "RateMoreProducts"
      val RATE_MORE_RECENTLY_PRODUCTS = "RateMoreRecentlyProducts"
      val AVERAGE_PRODUCTS = "AverageProducts"
    
      def main(args: Array[String]): Unit = { 
        val config = Map(
          "spark.cores" -> "local[*]",
          "mongo.uri" -> "mongodb://hadoop12:27017/recommender",
          "mongo.db" -> "recommender"
        )
    
        //创建一个spark config
        val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("DataLoader")
        //创建spark session
        val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    
        import spark.implicits._
        implicit val mongoConfig = MongoConfig(config("mongo.uri"),config("mongo.db"))
    
        //加载数据
        val ratingDF = spark.read
          .option("uri",mongoConfig.uri)
          .option("collection",MONGODB_RATING_COLLECTION)
          .format("com.mongodb.spark.sql")
          .load()
          .as[Rating]
          .toDF()
    
        //创建一张交ratings的临时表
        ratingDF.createOrReplaceTempView("ratings")
    
        //TODO :用spark sql去做不同的统计推荐
        //1.历史热门商品，按照评分个数统计,producetId,count
        val rateMoreProductsDF = spark.sql(
          """
            |select
            |productId,count(productId) as count
            |from ratings
            |group by productId
            |order by count desc
            |""".stripMargin)
        storeDFInMongoDB(rateMoreProductsDF,RATE_MORE_PRODUCTS)
    
        //2.近期热门商品，把时间戳转换成yyyyMM格式进行统计,最终得到productId，count，yearmonth
        //创建一个日期格式化工具
        val simpleDateFormat = new SimpleDateFormat("yyyyMM")
        //注册UDF，将timestamp转化为年月格式yyyyMM Date要long类型 format得到的是string 最后转化Int
        spark.udf.register("changeDate",(x:Int)=>simpleDateFormat.format(new Date(x * 1000L)).toInt)
        //把原始rating数据转换成想要的结构productId，score，yearmonth
        val ratingOfYearMonthDF = spark.sql(
          """
            |select
            |productId,socre,changeDate(timestamp) as yearmonth
            |from ratings
            |""".stripMargin)
        ratingOfYearMonthDF.createOrReplaceTempView("ratingOfMonth")
        val rateMoreRecentlyProductsDF = spark.sql(
          """
            |select
            |productId,count(productId) as count,yearmonth
            |from ratingOfMonth
            |group by yearmonth,productId
            |order by yearmonth desc,count desc
            |""".stripMargin)
        //把df保存到mongodb
        storeDFInMongoDB(rateMoreRecentlyProductsDF,RATE_MORE_RECENTLY_PRODUCTS)
    
        //3.优质商品统计，商品的平均评分 productId avg
        val averageProductsDF = spark.sql(
          """
            |select
            |productId,avg(socre) as avg
            |from ratings
            |group by productId
            |order by avg desc
            |""".stripMargin)
        storeDFInMongoDB(averageProductsDF,AVERAGE_PRODUCTS)
    
        spark.stop()
    
      }
      def storeDFInMongoDB(df:DataFrame, colllection_name: String)(implicit mongoConfig: MongoConfig): Unit ={ 
        df.write
          .option("uri",mongoConfig.uri)
          .option("collection",colllection_name)
          .mode("overwrite")
          .format("com.mongodb.spark.sql")
          .save()
      }
    
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]

# 二.扩展写到Mysql里 #

依赖

<dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>5.1.27</version>
    </dependency>

代码实现

import java.sql.Date
    import java.text.SimpleDateFormat
    
    import org.apache.spark.SparkConf
    import org.apache.spark.sql.{ DataFrame, SaveMode, SparkSession}
    
    case class Rating(userId:Int,productId:Int,score:Double,timestamp:Int)
    case class MysqlConfig(url:String,user:String,pw:String)
    
    object StatisticRecommender { 
      //定义mysqldb存储的表名
      val MYSQL_RATING_COLLECTION = "Rating"
    
      //统计的表的名称
      val RATE_MORE_PRODUCTS = "RateMoreProducts"
      val RATE_MORE_RECENTLY_PRODUCTS = "RateMoreRecentlyProducts"
      val AVERAGE_PRODUCTS = "AverageProducts"
    
      def main(args: Array[String]): Unit = { 
        val config = Map(
          "spark.cores" -> "local[*]",
          "mysql.url" -> "jdbc:mysql://hadoop12:3306/recommend?useUnicode=true&characterEncoding=UTF-8",
          "mysql.user" -> "root",
          "mysql.pw" -> "yy8266603"
        )
    
        //创建一个spark config
        val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("DataLoader")
        //创建spark session
        val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    
        import spark.implicits._
        implicit val mysqlConfig = MysqlConfig(config("mysql.url"),config("mysql.user"),config("mysql.pw"))
    
        //加载数据
        val ratingDF = spark.read
          .option("url",mysqlConfig.url)
          .option("user",mysqlConfig.user)
          .option("password",mysqlConfig.pw)
          .option("dbtable",MYSQL_RATING_COLLECTION)
          .format("jdbc")
          .load()
          .as[Rating]
          .toDF()
    
        //创建一张交ratings的临时表
        ratingDF.createOrReplaceTempView("ratings")
    
        //TODO :用spark sql去做不同的统计推荐
        //1.历史热门商品，按照评分个数统计,producetId,count
        val rateMoreProductsDF = spark.sql(
          """
            |select
            |productId,count(productId) as count
            |from ratings
            |group by productId
            |order by count desc
            |""".stripMargin)
        storeDFInMysql(rateMoreProductsDF,RATE_MORE_PRODUCTS)
    
        //2.近期热门商品，把时间戳转换成yyyyMM格式进行统计,最终得到productId，count，yearmonth
        //创建一个日期格式化工具
        val simpleDateFormat = new SimpleDateFormat("yyyyMM")
        //注册UDF，将timestamp转化为年月格式yyyyMM Date要long类型 format得到的是string 最后转化Int
        spark.udf.register("changeDate",(x:Int)=>simpleDateFormat.format(new Date(x * 1000L)).toInt)
        //把原始rating数据转换成想要的结构productId，score，yearmonth
        val ratingOfYearMonthDF = spark.sql(
          """
            |select
            |productId,score,changeDate(timestamp) as yearmonth
            |from ratings
            |""".stripMargin)
        ratingOfYearMonthDF.createOrReplaceTempView("ratingOfMonth")
        val rateMoreRecentlyProductsDF = spark.sql(
          """
            |select
            |productId,count(productId) as count,yearmonth
            |from ratingOfMonth
            |group by yearmonth,productId
            |order by yearmonth desc,count desc
            |""".stripMargin)
        //把df保存到mongodb
        storeDFInMysql(rateMoreRecentlyProductsDF,RATE_MORE_RECENTLY_PRODUCTS)
    
        //3.优质商品统计，商品的平均评分 productId avg
        val averageProductsDF = spark.sql(
          """
            |select
            |productId,avg(score) as avg
            |from ratings
            |group by productId
            |order by avg desc
            |""".stripMargin)
        storeDFInMysql(averageProductsDF,AVERAGE_PRODUCTS)
    
        spark.stop()
    
      }
      def storeDFInMysql(df:DataFrame,dbtable_name:String)(implicit mysqlConfig: MysqlConfig): Unit ={ 
        df.write
          .option("url",mysqlConfig.url)
          .option("user",mysqlConfig.user)
          .option("password",mysqlConfig.pw)
          .option("dbtable",dbtable_name)
          .mode(SaveMode.Overwrite)
          .format("jdbc")
          .save()
    
      }
    
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/eee0a95045a84a2baf088856fb7a1009.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/2c70577323794957984554d99c177814.png