人工智能-推荐系统-模块01：离线统计模块【使用SparkSQL（基于Scala语言/Python语言）进行离线统计分析：历史热门商品统计、近期热门商品统计、商品平均评分统计...】

￡神魔★判官ぃ 2023-10-05 18:49 2阅读 0赞

## 一、基于scala语言的SparkSQL离线统计分析 ##

### 1、将数据导入MongoDB数据库 ###

DataLoader.scala

import com.mongodb.casbah.commons.MongoDBObject
    import com.mongodb.casbah.{
        MongoClient, MongoClientURI}
    import org.apache.spark.SparkConf
    import org.apache.spark.sql.{
        DataFrame, SparkSession}
    
    /**
      * Project: ECommerceRecommendSystem
      * Created on 2019/4/26 15:43
      */
    
    /**
      * Product数据集
      * 3982                            商品ID
      * Fuhlen 富勒 M8眩光舞者时尚节能    商品名称
      * 1057,439,736                    商品分类ID，不需要
      * B009EJN4T2                      亚马逊ID，不需要
      * https://images-cn-4.ssl-image   商品的图片URL
      * 外设产品|鼠标|电脑/办公           商品分类
      * 富勒|鼠标|电子产品|好用|外观漂亮   商品UGC标签
      */
    case class Product( productId: Int, name: String, imageUrl: String, categories: String, tags: String )
    
    /**
      * Rating数据集
      * 4867        用户ID
      * 457976      商品ID
      * 5.0         评分
      * 1395676800  时间戳
      */
    case class Rating( userId: Int, productId: Int, score: Double, timestamp: Int )
    
    /**
      * MongoDB连接配置
      * @param uri    MongoDB的连接uri
      * @param db     要操作的db
      */
    case class MongoConfig( uri: String, db: String )
    
    object DataLoader {
        
      // 定义数据文件路径
      val PRODUCT_DATA_PATH = "D:\\Projects\\BigData\\ECommerceRecommendSystem\\recommender\\DataLoader\\src\\main\\resources\\products.csv"
      val RATING_DATA_PATH = "D:\\Projects\\BigData\\ECommerceRecommendSystem\\recommender\\DataLoader\\src\\main\\resources\\ratings.csv"
      // 定义mongodb中存储的表名
      val MONGODB_PRODUCT_COLLECTION = "Product"
      val MONGODB_RATING_COLLECTION = "Rating"
    
      def main(args: Array[String]): Unit = {
        
        val config = Map(
          "spark.cores" -> "local[*]",
          "mongo.uri" -> "mongodb://localhost: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._
    
        // 加载数据
        val productRDD = spark.sparkContext.textFile(PRODUCT_DATA_PATH)
        val productDF = productRDD.map( item => {
        
          // product数据通过^分隔，切分出来
          val attr = item.split("\\^")
          // 转换成Product
          Product( attr(0).toInt, attr(1).trim, attr(4).trim, attr(5).trim, attr(6).trim )
        } ).toDF()
    
        val ratingRDD = spark.sparkContext.textFile(RATING_DATA_PATH)
        val ratingDF = ratingRDD.map( item => {
        
          val attr = item.split(",")
          Rating( attr(0).toInt, attr(1).toInt, attr(2).toDouble, attr(3).toInt )
        } ).toDF()
    
        implicit val mongoConfig = MongoConfig( config("mongo.uri"), config("mongo.db") )
        storeDataInMongoDB( productDF, ratingDF )
    
        spark.stop()
      }
      def storeDataInMongoDB( productDF: DataFrame, ratingDF: DataFrame )(implicit mongoConfig: MongoConfig): Unit ={
        
        // 新建一个mongodb的连接，客户端
        val mongoClient = MongoClient( MongoClientURI(mongoConfig.uri) )
        // 定义要操作的mongodb表，可以理解为 db.Product
        val productCollection = mongoClient( mongoConfig.db )( MONGODB_PRODUCT_COLLECTION )
        val ratingCollection = mongoClient( mongoConfig.db )( MONGODB_RATING_COLLECTION )
    
        // 如果表已经存在，则删掉
        productCollection.dropCollection()
        ratingCollection.dropCollection()
    
        // 将当前数据存入对应的表中
        productDF.write
          .option("uri", mongoConfig.uri)
          .option("collection", MONGODB_PRODUCT_COLLECTION)
          .mode("overwrite")
          .format("com.mongodb.spark.sql")
          .save()
    
        ratingDF.write
          .option("uri", mongoConfig.uri)
          .option("collection", MONGODB_RATING_COLLECTION)
          .mode("overwrite")
          .format("com.mongodb.spark.sql")
          .save()
    
        // 对表创建索引
        productCollection.createIndex( MongoDBObject( "productId" -> 1 ) )
        ratingCollection.createIndex( MongoDBObject( "productId" -> 1 ) )
        ratingCollection.createIndex( MongoDBObject( "userId" -> 1 ) )
    
        mongoClient.close()
      }
    }

### 2、使用SparkSQL分析MongoDB数据库数据进行离线统计分析 ###

StatisticsRecommender.scala

import java.text.SimpleDateFormat
    import java.util.Date
    
    import org.apache.spark.SparkConf
    import org.apache.spark.sql.{
        DataFrame, SparkSession}
    
    /**
      * Project: ECommerceRecommendSystem
      * Created on 2019/4/27 8:53
      */
    case class Rating( userId: Int, productId: Int, score: Double, timestamp: Int )
    case class MongoConfig( uri: String, db: String )
    
    object StatisticsRecommender {
        
      // 定义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[1]",
          "mongo.uri" -> "mongodb://localhost:27017/recommender",
          "mongo.db" -> "recommender"
        )
        // 创建一个spark config
        val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("StatisticsRecommender")
        // 创建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")
    
        /**
          * 用spark sql去做不同的统计推荐
          */
        // 1. 历史热门商品，按照评分个数统计，productId，count
        val rateMoreProductsDF = spark.sql("select productId, count(productId) as count from ratings group by productId order by count desc")
        storeDFInMongoDB( rateMoreProductsDF, RATE_MORE_PRODUCTS )
    
        // 2. 近期热门商品，把时间戳转换成yyyyMM格式进行评分个数统计，最终得到productId, count, yearmonth
        // 创建一个日期格式化工具
        val simpleDateFormat = new SimpleDateFormat("yyyyMM")
        // 注册UDF，将timestamp转化为年月格式yyyyMM
        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")
        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")
        // 把df保存到mongodb
        storeDFInMongoDB( 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")
        storeDFInMongoDB( averageProductsDF, AVERAGE_PRODUCTS )
    
        spark.stop()
      }
      def storeDFInMongoDB(df: DataFrame, collection_name: String)(implicit mongoConfig: MongoConfig): Unit ={
        
        df.write
          .option("uri", mongoConfig.uri)
          .option("collection", collection_name)
          .mode("overwrite")
          .format("com.mongodb.spark.sql")
          .save()
      }
    }

## 二、基于Python语言的SparkSQL离线统计分析 ##