电商推荐系统三：创建项目并初始化业务数据

柔情只为你懂 2022-10-28 10:16 179阅读 0赞

# 三、创建项目并初始化业务数据 #

### 目录 ###

*  三、创建项目并初始化业务数据
 *   *  3.1 在IDEA中创建maven项目
     *   *  3.1.1 项目框架搭建
         *  3.1.2 声明项目中工具的版本信息
         *  3.1.3 添加项目依赖
     *  3.2 数据加载准备
     *   *  3.2.1 Products数据集
         *  3.2.2 Ratings数据集
         *  3.2.3 日志管理配置文件
     *  3.3 数据初始化到MongoDB
     *   *  3.3.1 启动MongoDB数据库（略）
         *  3.3.2 数据加载程序主体实现
         *  3.3.3 将数据写入MongoDB
     *  3.4 附件：完整代码

我们的项目主体用Scala编写，采用IDEA作为开发环境进行项目编写，采用maven作为项目构建和管理工具。

## 3.1 在IDEA中创建maven项目 ##

打开IDEA，创建一个maven项目，命名为ECommerceRecommendSystem。为了方便后期的联调，我们会把业务系统的代码也添加进来，所以我们可以以ECommerceRecommendSystem作为父项目，并在其下建一个名为recommender的子项目，然后再在下面搭建多个子项目用于提供不同的推荐服务。

### 3.1.1 项目框架搭建 ###

在ECommerceRecommendSystem下新建一个 maven module作为子项目，命名为recommender。同样的，再以recommender为父项目，新建一个maven module作为子项目。我们的第一步是初始化业务数据，所以子项目命名为 DataLoader。

父项目只是为了规范化项目结构，方便依赖管理，本身是不需要代码实现的，所以ECommerceRecommendSystem和recommender下的src文件夹都可以删掉。

目前的整体项目框架如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

### 3.1.2 声明项目中工具的版本信息 ###

我们整个项目需要用到多个工具，它们的不同版本可能会对程序运行造成影响，所以应该在最外层的ECommerceRecommendSystem中声明所有子项目共用的版本信息。

在pom.xml中加入以下配置：ECommerceRecommendSystem/pom.xml

### 3.1.3 添加项目依赖 ###

首先，对于整个项目而言，应该有同样的日志管理，我们在ECommerceRecommendSystem中引入公有依赖：

ECommerceRecommendSystem/pom.xml

<dependencies>
        
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>jcl-over-slf4j</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>${log4j.version}</version>
        </dependency>
    </dependencies>

同样，对于maven项目的构建，可以引入公有的插件：

<build>
        
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.6.1</version>
                
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
        <pluginManagement>
            <plugins>
                
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-assembly-plugin</artifactId>
                    <version>3.0.0</version>
                    <executions>
                        <execution>
                            <id>make-assembly</id>
                            <phase>package</phase>
                            <goals>
                                <goal>single</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
                
                <plugin>
                    <groupId>net.alchim31.maven</groupId>
                    <artifactId>scala-maven-plugin</artifactId>
                    <version>3.2.2</version>
                    <executions>
                        
                        <execution>
                            <goals>
                                <goal>compile</goal>
                                <goal>testCompile</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </pluginManagement>
    </build>

然后，在recommender模块中，我们可以为所有的推荐模块声明spark相关依赖（这里的dependencyManagement表示仅声明相关信息，子项目如果依赖需要自行引入）：

ECommerceRecommendSystem/recommender/pom.xml

<dependencyManagement>
        <dependencies>
            
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-core_2.11</artifactId>
                <version>${spark.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-sql_2.11</artifactId>
                <version>${spark.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming_2.11</artifactId>
                <version>${spark.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-mllib_2.11</artifactId>
                <version>${spark.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-graphx_2.11</artifactId>
                <version>${spark.version}</version>
            </dependency>
    <dependency>
                <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>${scala.version}</version>
    </dependency>
        </dependencies>
    </dependencyManagement>

由于各推荐模块都是scala代码，还应该引入scala-maven-plugin插件，用于scala程序的编译。因为插件已经在父项目中声明，所以这里不需要再声明版本和具体配置：

<build>
        <plugins>
            
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

对于具体的DataLoader子项目，需要spark相关组件，还需要mongodb的相关依赖，我们在pom.xml文件中引入所有依赖（在父项目中已声明的不需要再加详细信息）：  
ECommerceRecommendSystem/recommender/DataLoader/pom.xml

<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>

至此，我们做数据加载需要的依赖都已配置好，可以开始写代码了。

## 3.2 数据加载准备 ##

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

### 3.2.1 Products数据集 ###

数据格式：

productId,name,categoryIds, amazonId, imageUrl, categories, tags

例如：

3982^Fuhlen 富勒 M8眩光舞者时尚节能无线鼠标(草绿)(眩光.悦动.时尚炫舞鼠标 12个月免换电池 高精度光学寻迹引擎 超细微接收器10米
    传输距离)^1057,439,736^B009EJN4T2^https://images-cn-4.ssl-images-amazon.com/images/I/31QPvUDNavL._SY300_QL70
    _.jpg^外设产品|鼠标|电脑/办公^富勒|鼠标|电子产品|好用|外观漂亮

Product数据集有7个字段，每个字段之间通过“^”符号进行分割。其中的categoryIds、amazonId对于内容特征没有实质帮助，我们只需要其它5个字段：

<table> 
 <thead> 
  <tr> 
   <th>字段名</th> 
   <th>字段类型</th> 
   <th>字段描述</th> 
   <th>字段备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>productId</td> 
   <td>Int</td> 
   <td>商品ID</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>name</td> 
   <td>String</td> 
   <td>商品名称</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>categories</td> 
   <td>String</td> 
   <td>商品分类</td> 
   <td>每一项用“|”分割</td> 
  </tr> 
  <tr> 
   <td>imageUrl</td> 
   <td>String</td> 
   <td>商品图片URL</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>tags</td> 
   <td>String</td> 
   <td>商品UGC标签</td> 
   <td>每一项用“|”分割</td> 
  </tr> 
 </tbody> 
</table>

### 3.2.2 Ratings数据集 ###

数据格式：

userId,prudcutId,rating,timestamp

例如：

4867,457976,5.0,1395676800

Rating数据集有4个字段，每个字段之间通过“，”分割。

<table> 
 <thead> 
  <tr> 
   <th>字段名</th> 
   <th>字段类型</th> 
   <th>字段描述</th> 
   <th>字段备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>userId</td> 
   <td>Int</td> 
   <td>用户ID</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>produtId</td> 
   <td>Int</td> 
   <td>商品ID</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>score</td> 
   <td>Double</td> 
   <td>评分值</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>timestamp</td> 
   <td>Long</td> 
   <td>评分的时间</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

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

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

## 3.3 数据初始化到MongoDB ##

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

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

我们会为原始数据定义几个样例类，通过SparkContext的textFile方法从文件中读取数据，并转换成DataFrame，再利用Spark SQL提供的write方法进行数据的分布式插入。

在DataLoader/src/main/scala下新建package，命名为com.atguigu.recommender，新建名为DataLoader的scala class文件。

程序主体代码如下：

DataLoader/src/main/scala/com.recom.recommerder/DataLoader.scala

// 定义样例类
    case class Product(productId: Int, name: String, imageUrl: String, categories: String, 
    tags: String)
    case class Rating(userId: Int, productId: Int, score: Double, timestamp: Int)
    
    case class MongoConfig(uri:String, db:String)
    
    object DataLoader { 
      // 以window下为例，需替换成自己的路径，linux下为 /YOUR_PATH/resources/products.csv
      val PRODUCT_DATA_PATH = " YOUR_PATH\\resources\\products.csv"
      val RATING_DATA_PATH = " YOUR_PATH\\resources\\ratings.csv"
    
      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"
        )
        // 创建一个SparkConf配置
        val sparkConf = new
    SparkConf().setAppName("DataLoader").setMaster(config("spark.cores"))
        // 创建一个SparkSession 
        val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    
        // 在对DataFrame和Dataset进行操作许多操作都需要这个包进行支持
        import spark.implicits._
    
        // 将Product、Rating数据集加载进来
        val productRDD = spark.sparkContext.textFile(PRODUCT_DATA_PATH)
          //将ProdcutRDD装换为DataFrame
        val productDF = productRDD.map(item =>{ 
            val attr = item.split("\\^")
    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)
          //将ratingRDD转换为DataFrame
        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.get("mongo.uri").get,config.get("mongo.db").get)
        // 将数据保存到MongoDB中
        storeDataInMongoDB(productDF, ratingDF)
    
        // 关闭Spark
        spark.stop()
      }

### 3.3.3 将数据写入MongoDB ###

接下来，实现storeDataInMongo方法，将数据写入mongodb中：

def storeDataInMongoDB(productDF: DataFrame, ratingDF:DataFrame)
    	(implicit mongoConfig: MongoConfig): Unit = { 
    
      //新建一个到MongoDB的连接
      val mongoClient = MongoClient(MongoClientURI(mongoConfig.uri))
    
      // 定义通过MongoDB客户端拿到的表操作对象
      val productCollection = mongoClient(mongoConfig.db)(MONGODB_PRODUCT_COLLECTION)
    val ratingCollection = mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION)
    
      //如果MongoDB中有对应的数据库，那么应该删除
      productCollection.dropCollection()
      ratingCollection.dropCollection()
    
      //将当前数据写入到MongoDB
      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("userId" -> 1))
    ratingCollection.createIndex(MongoDBObject("productId" -> 1))
    
      //关闭MongoDB的连接
      mongoClient.close()
    }

## 3.4 附件：完整代码 ##

package com.recom.dataload
    
    import com.mongodb.casbah.commons.MongoDBObject
    import com.mongodb.casbah.{ MongoClient, MongoClientURI, MongoCollection}
    import org.apache.spark.SparkConf
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.{ DataFrame, SparkSession}
    
    
    /**
     * 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:\\IdeaProject\\gmallRecommend\\recommender\\DataLoader\\src\\main\\resources\\products.csv"
      val RATING_DATA_PATH="D:\\IdeaProject\\gmallRecommend\\recommender\\DataLoader\\src\\main\\resources\\ratings.csv"
      
      //定义monogobd中存储的表名
      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://hadoop102: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()
        
        //导入隐式转换类，在DF和DS转换的过程中会使用到
        import spark.implicits._
        
        //加载数据
        val productRDD: RDD[String] = spark.sparkContext.textFile(PRODUCT_DATA_PATH)
        val productDF: DataFrame = productRDD.map(item => { 
          val attr: Array[String] = item.split("\\^")
          //封装为Product类
          Product(attr(0).toInt, attr(1).trim, attr(4).trim, attr(5).trim, attr(6).trim)
        }).toDF()
       
        val ratingRDD: RDD[String] = spark.sparkContext.textFile(RATING_DATA_PATH)
        val ratingDF: DataFrame = ratingRDD.map(item => { 
          val attr: Array[String] = item.split(",")
          //封装数据
          Rating(attr(0).toInt, attr(1).toInt, attr(2).toDouble, attr(3).toInt)
        }).toDF()
    
        
        //通过隐式类的方法创建mongodb连接对象
        implicit val mongoConfig = MongoConfig(config("mongo.uri"),config("mongo.db"))
        //将数据加载到mongodb
        storeInMongoDB(productDF,ratingDF)
        
        spark.stop()
      }
      
      //使用隐式传参的方法，减少数据的反复传递
      def storeInMongoDB(productDF:DataFrame,ratingDF:DataFrame)(implicit mongoConfig: MongoConfig)={ 
        
        //新建一个mongodb的客户端连接
        val mongoClient: MongoClient = MongoClient(MongoClientURI(mongoConfig.uri))
        //定义要做操作的表
        val productCollection: MongoCollection = mongoClient(mongoConfig.db)(MONGODB_PRODUCT_COLLECTION)
        val ratingCollection: MongoCollection = 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()
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221024/a3abfb10027c4811b89d1c4cb48ecd87.png