大数据技术之电商推荐系统(4) | 初始化业务数据

女爷i 2022-12-21 03:08 178阅读 0赞

### 目录 ###

*  一.数据加载准备
 *   *  Products数据集
     *  Ratings数据集
     *  日志管理配置文件
 *  二.数据初始化到MongoDB
 *   *  启动MongoDB数据库（略）
     *  数据加载程序主体实现
 *  三.扩展写到Mysql里

# 一.数据加载准备 #

在src/main/目录下，可以看到已有的默认源文件目录是java，我们可以将其改名为scala。将数据文products.csv，ratings.csv复制到资源文件目录src/main/resources下，我们将从这里读取数据并加载到mongodb中。  
**数据**

链接：https://pan.baidu.com/s/1eYfyuzhb0MbTtcGX2okcxg 
    提取码：5vd5

## Products数据集 ##

数据格式：  
productId,name,categoryIds, amazonId, imageUrl, categories, tags

*  Product数据集有7个字段，每个字段之间通过“^”符号进行分割。其中的categoryIds、amazonId对于内容特征没有实质帮助，我们只需要其它5个字段：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]

## Ratings数据集 ##

数据格式：  
userId,prudcutId,rating,timestamp  
例如：  
4867,457976,5.0,1395676800  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]

## 日志管理配置文件 ##

log4j对日志的管理，需要通过配置文件来生效。在src/main/resources下新建配置文件log4j.properties，写入以下内容：

log4j.rootLogger=info, 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

# 二.数据初始化到MongoDB #

## 启动MongoDB数据库（略） ##

## 数据加载程序主体实现 ##

*  我们会为原始数据定义几个样例类，通过SparkContext的textFile方法从文件中读取数据，并转换成DataFrame，再利用Spark SQL提供的write方法进行数据的分布式插入。
 *  在DataLoader/src/main/scala下新建package，命名为com.donglin.recommender，新建名为DataLoader的scala class文件。  
    **程序主体代码如下：**

import com.mongodb.casbah.commons.MongoDBObject
    import com.mongodb.casbah.{ MongoClient, MongoClientURI}
    import org.apache.spark.SparkConf
    import org.apache.spark.sql.{ DataFrame, SparkSession}
    
    
    /** * 3982 商品ID * Fuhlen富勒 M 商品名称 * 1057,439,736 商品分类ID x * B009EJN4T2 亚马逊ID x * https://images-cn-4 商品的图片URL * 外设产品|鼠标|电脑/办公 商品分类 * 富勒|鼠标|电子产品| 商品UGC标签 */
    case class Product(productId:Int,name :String,imageUrl:String,categories:String,tags:String)
    
    /** * 4867 用户ID * 457976 商品ID * 5.0 评分 * 1395676800 时间戳 */
    case class Rating(userId:Int,productId:Int,socre:Double,timestamp:Int)
    
    //MongoDB的连接配置
    case class MongoConfig(uri:String,db:String)
    
    object DataLoader { 
      //定义数据文件路径
      val RRODUCT_DATA_PATH = "D:\\idea\\ECommerceRecommendSystem\\recommender\\DataLoader\\src\\main\\resources\\products.csv"
      val RATING_DATA_PATH = "D:\\idea\\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 conf = Map(
          "spark.cores" -> "local[*]",
          "mongo.url" -> "mongodb://hadoop12:27017/recommender",
          "mongo.db" -> "recommender"
        )
    
        //创建一个spark config
        val sparkConf = new SparkConf().setMaster(conf("spark.cores")).setAppName("DataLoader")
        //创建spark session
        val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    
        import spark.implicits._
    
        //加载数据
        val productRDD = spark.sparkContext.textFile(RRODUCT_DATA_PATH)
        val productDF = productRDD.map(item =>{ 
          //product通过^分隔，切分出来
          val dataArray = item.split("\\^")
          //转换成Product
          Product(dataArray(0).trim.toInt,dataArray(1).trim,dataArray(4).trim,dataArray(5).trim,dataArray(6).trim)
        }).toDF()
    
        val ratingRDD = spark.sparkContext.textFile(RATING_DATA_PATH)
        val ratingDF = ratingRDD.map(item =>{ 
          val dataArray = item.split(",")
          Rating(dataArray(0).trim.toInt,dataArray(1).trim.toInt,dataArray(2).trim.toDouble,dataArray(3).trim.toInt)
        }).toDF()
    
        implicit val mongoConfig = MongoConfig( conf("mongo.url"), conf("mongo.db"))
        storeDataInMogoDB(productDF,ratingDF)
    
        spark.stop()
      }
    
      def storeDataInMogoDB(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()
    
      }
    
    }

使用mongodb查看结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 3]

# 三.扩展写到Mysql里 #

依赖

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

代码实现

import org.apache.spark.SparkConf
    import org.apache.spark.sql.{ DataFrame, SaveMode, SparkSession}
    
    
    /** * 3982 商品ID * Fuhlen富勒 M 商品名称 * 1057,439,736 商品分类ID x * B009EJN4T2 亚马逊ID x * https://images-cn-4 商品的图片URL * 外设产品|鼠标|电脑/办公 商品分类 * 富勒|鼠标|电子产品| 商品UGC标签 */
    case class Product(productId:Int,name :String,imageUrl:String,categories:String,tags:String)
    
    /** * 4867 用户ID * 457976 商品ID * 5.0 评分 * 1395676800 时间戳 */
    case class Rating(userId:Int,productId:Int,score:Double,timestamp:Int)
    
    //Mysql的连接配置
    case class MysqlConfig(url:String,user:String,pw:String)
    
    object DataLoader { 
      //定义数据文件路径
      val RRODUCT_DATA_PATH = "D:\\idea\\ECommerceRecommendSystem\\recommender\\DataLoader\\src\\main\\resources\\products.csv"
      val RATING_DATA_PATH = "D:\\idea\\ECommerceRecommendSystem\\recommender\\DataLoader\\src\\main\\resources\\ratings.csv"
      //定义mongodb存储的表名
      val MYSQL_PRODUCT_COLLECTION = "Product"
      val MYSQL_RATING_COLLECTION = "Rating"
    
      def main(args: Array[String]): Unit = { 
        val conf = 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(conf("spark.cores")).setAppName("DataLoader")
        //创建spark session
        val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    
        import spark.implicits._
    
        //加载数据
        val productRDD = spark.sparkContext.textFile(RRODUCT_DATA_PATH)
        val productDF = productRDD.map(item =>{ 
          //product通过^分隔，切分出来
          val dataArray = item.split("\\^")
          //转换成Product
          Product(dataArray(0).trim.toInt,dataArray(1).trim,dataArray(4).trim,dataArray(5).trim,dataArray(6).trim)
        }).toDF()
    
        val ratingRDD = spark.sparkContext.textFile(RATING_DATA_PATH)
        val ratingDF = ratingRDD.map(item =>{ 
          val dataArray = item.split(",")
          Rating(dataArray(0).trim.toInt,dataArray(1).trim.toInt,dataArray(2).trim.toDouble,dataArray(3).trim.toInt)
        }).toDF()
    
        implicit val mysqlConfig = MysqlConfig(conf("mysql.url"),conf("mysql.user"),conf("mysql.pw"))
        storeDataInMysql(productDF,ratingDF)
    
        spark.stop()
      }
    
      def storeDataInMysql(productDF:DataFrame,ratingDF:DataFrame)(implicit mysqlConfig: MysqlConfig): Unit ={ 
    
        //将当前数据存入对应的表中
        productDF.write
          .option("url",mysqlConfig.url)
          .option("user",mysqlConfig.user)
          .option("password",mysqlConfig.pw)
          .option("dbtable",MYSQL_PRODUCT_COLLECTION)
          .mode(SaveMode.Overwrite)
          .format("jdbc")
          .save()
    
        ratingDF.write
          .option("url",mysqlConfig.url)
          .option("user",mysqlConfig.user)
          .option("password",mysqlConfig.pw)
          .option("dbtable",MYSQL_RATING_COLLECTION)
          .mode(SaveMode.Overwrite)
          .format("jdbc")
          .save()
    
      }
    
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/524f46b888d246e9ac340360fb4be3f0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/32a41de7ec834eb8850daeac773d598f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/15ced34e8e9544ff9a2540347dde881c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221120/21d9a5016cd645fcaca9b289412f8b4c.png