大数据技术之_24_电影推荐系统项目_06_项目体系架构设计 + 工具环境搭建 + 创建项目并初始化业务数据 + 离线推荐服务建设 + 实时推荐服务建设 + 基于内容的推荐服务建设

    <properties>
        <log4j.version>1.2.17</log4j.version><!-- log4j 日志系统 -->
        <!-- SLF4J，即简单日志门面（Simple Logging Facade for Java），不是具体的日志解决方案，它只服务于各种各样的日志系统。
        按照官方的说法，SLF4J 是一个用于日志系统的简单Facade，允许最终用户在部署其应用时使用其所希望的日志系统。 -->
        <slf4j.version>1.7.22</slf4j.version>
        <mongodb-spark.version>2.0.0</mongodb-spark.version>
        <casbah.version>3.1.1</casbah.version><!-- mongodb 在 scala 上的驱动器 -->
        <elasticsearch-spark.version>5.6.2</elasticsearch-spark.version>
        <elasticsearch.version>5.6.2</elasticsearch.version>
        <redis.version>2.9.0</redis.version>
        <kafka.version>0.10.2.1</kafka.version>
        <spark.version>2.1.1</spark.version>
        <scala.version>2.11.8</scala.version>
        <jblas.version>1.2.1</jblas.version><!-- Java 中线性代数相关的库 -->
    </properties>

    <dependencies>
        <!-- 引入共同的日志管理工具 -->
        <!-- https://mvnrepository.com/artifact/log4j/log4j -->
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>${log4j.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.slf4j/jcl-over-slf4j -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>jcl-over-slf4j</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-log4j12 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
    </dependencies>

    <build>
        <!-- 声明并引入子项目共有的插件 -->
        <plugins>
            <!-- 直接引入插件，所有子模块都会引入 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.6.1</version>
                <!-- 所有的编译用 JDK1.8 -->
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>

        <pluginManagement>
            <!-- 只是声明插件，并不引入，如果子模块需要用的时候再引入即可 -->
            <plugins>
                <!-- maven 的打包插件 -->
                <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>
                <!-- 该插件用于将 scala 代码编译成 class 文件 -->
                <plugin>
                    <groupId>net.alchim31.maven</groupId>
                    <artifactId>scala-maven-plugin</artifactId>
                    <version>3.2.2</version>
                    <executions>
                        <!-- 绑定到 maven 的编译阶段 -->
                        <execution>
                            <goals>
                                <goal>compile</goal>
                                <goal>testCompile</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </pluginManagement>
    </build>

    <dependencyManagement>
        <dependencies>
            <!-- 引入 Spark 相关的 Jar 包 -->
            <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>

    <build>
        <plugins>
            <!-- 父项目已声明该 plugin，子项目在引入的时候，不用声明版本和已经声明的配置 -->
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

    <dependencies>
        <!-- Spark 的依赖引入 -->
        <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>
        <!-- 引入 Scala -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
        </dependency>
        <!-- 加入 MongoDB 的驱动 -->
        <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>
        <!-- 加入 ElasticSearch 的驱动 -->
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>transport</artifactId>
            <version>${elasticsearch.version}</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch-spark-20_2.11</artifactId>
            <version>${elasticsearch-spark.version}</version>
            <!-- 将不需要依赖的包从依赖路径中除去 -->
            <exclusions>
                <exclusion>
                    <groupId>org.apache.hive</groupId>
                    <artifactId>hive-service</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>

mid,name,descri,timelong,issue,shoot,language,genres,actors,directors

e.g.
1^Toy Story (1995)^ ^81 minutes^March 20, 2001^1995^English
^Adventure|Animation|Children|Comedy|Fantasy ^Tom Hanks|Tim Allen|Don Rickles|Jim Varney|Wallace Shawn|John Ratzenberger|Annie Potts|John Morris|Erik von Detten|Laurie Metcalf|R. Lee Ermey|Sarah Freeman|Penn Jillette|Tom Hanks|Tim Allen|Don Rickles|Jim Varney|Wallace Shawn ^John Lasseter

uid,mid,score,timestamp

e.g.
1,31,2.5,1260759144

uid,mid,tag,timestamp

e.g.
1,31,action,1260759144

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

package com.atguigu.recommender

import java.net.InetAddress

import com.mongodb.casbah.commons.MongoDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest
import org.elasticsearch.common.settings.Settings
import org.elasticsearch.common.transport.InetSocketTransportAddress
import org.elasticsearch.transport.client.PreBuiltTransportClient

/**
  * Movie 数据集
  *
  * 260                                         电影ID，mid
  * Star Wars: Episode IV - A New Hope (1977)   电影名称，name
  * Princess Leia is captured and held hostage  详情描述，descri
  * 121 minutes                                 时长，timelong
  * September 21, 2004                          发行时间，issue
  * 1977                                        拍摄时间，shoot
  * English                                     语言，language
  * Action|Adventure|Sci-Fi                     类型，genres
  * Mark Hamill|Harrison Ford|Carrie Fisher     演员表，actors
  * George Lucas                                导演，directors
  */
case class Movie(mid: Int, name: String, descri: String, timelong: String, issue: String,
                 shoot: String, language: String, genres: String, actors: String, directors: String)

/**
  * Rating 数据集
  *
  * 1,31,2.5,1260759144
  */
case class Rating(uid: Int, mid: Int, score: Double, timestamp: Long)

/**
  * Tag 数据集
  *
  * 15,1955,dentist,1193435061
  */
case class Tag(uid: Int, mid: Int, tag: String, timestamp: Long)

// 把 MongoDB 和 Elasticsearch 的配置封装成样例类
/**
  * @param uri MongDB 的连接
  * @param db  MongDB 的 数据库
  */
case class MongoConfig(uri: String, db: String)

/**
  * @param httpHosts      ES 的 http 主机列表，逗号分隔
  * @param transportHosts ES 的 http 端口列表，逗号分隔
  * @param index          需要操作的索引库，即数据库
  * @param clusterName    集群名称：默认是 my-application
  */
case class ESConfig(httpHosts: String, transportHosts: String, index: String, clusterName: String)

object DataLoader {

  // 定义常量
  // 以 Window 下为例，需替换成自己的路径，linux 下为 /YOUR_PATH/resources/movies.csv
  val MOVIE_DATA_PATH = "D:\\learn\\JetBrains\\workspace_idea\\MovieRecommendSystem\\recommender\\DataLoader\\src\\main\\resources\\movies.csv"
  val TATING_DATA_PATH = "D:\\learn\\JetBrains\\workspace_idea\\MovieRecommendSystem\\recommender\\DataLoader\\src\\main\\resources\\ratings.csv"
  val TAG_DATA_PATH = "D:\\learn\\JetBrains\\workspace_idea\\MovieRecommendSystem\\recommender\\DataLoader\\src\\main\\resources\\tags.csv"

  // 定义 MongoDB 数据库中的一些表名
  val MONGODB_MOVIE_COLLECTION = "Movie"
  val MONGODB_RATING_COLLECTION = "Rating"
  val MONGODB_TAG_COLLECTION = "Tag"

  // 定义 ES 中的一些索引（即数据库）
  val ES_MOVIE_INDEX = "Movie"

  // 主程序的入口
  def main(args: Array[String]): Unit = {
    // 定义用到的配置参数
    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://localhost:27017/recommender",
      "mongo.db" -> "recommender",
      "es.httpHosts" -> "hadoop102:9200",
      "es.transportHosts" -> "hadoop102:9300",
      "es.index" -> "recommender",
      "es.cluster.name" -> "my-application"
    )

    // 创建一个 SparkConf 对象
    val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("DataLoader")

    // 创建一个 SparkSession 对象
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    // 在对 DataFrame 和 Dataset 进行许多操作都需要这个包进行支持
    import spark.implicits._

    // 加载数据，将 Movie、Rating、Tag 数据集加载进来
    // 数据预处理
    val movieRDD = spark.sparkContext.textFile(MOVIE_DATA_PATH)
    // 将 movieRDD 转换为 DataFrame
    val movieDF = movieRDD.map(
      item => {
        val attr = item.split("\\^")
        Movie(attr(0).toInt, attr(1).trim, attr(2).trim, attr(3).trim, attr(4).trim, attr(5).trim, attr(6).trim, attr(7).trim, attr(8).trim, attr(9).trim)
      }
    ).toDF()

    val ratingRDD = spark.sparkContext.textFile(TATING_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).toLong)
      }
    ).toDF()

    val tagRDD = spark.sparkContext.textFile(TAG_DATA_PATH)
    // 将 tagRDD 装换为 DataFrame
    val tagDF = tagRDD.map(
      item => {
        val attr = item.split(",")
        Tag(attr(0).toInt, attr(1).toInt, attr(2).trim, attr(3).toLong)
      }
    ).toDF()

    // 声明一个隐式的配置对象
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))
    // 将数据保存到 MongoDB 中
    storeDataInMongDB(movieDF, ratingDF, tagDF)

    import org.apache.spark.sql.functions._
    // 数据预处理，把 movie 对应的 tag 信息添加进去，加一列，使用 “|” 分隔：tag1|tag2|...
    /**
      * mid,tags
      * tags: tag1|tag2|tag3|...
      */
    val newTag = tagDF.groupBy($"mid")
      .agg(concat_ws("|", collect_set($"tag")).as("tags")) // groupby 为对原 DataFrame 进行打包分组，agg 为聚合(其操作包括 max、min、std、sum、count)
      .select("mid", "tags")
    // 将 movie 和 newTag 作 左外连接，把数据合在一起
    val movieWithTagsDF = movieDF.join(newTag, Seq("mid"), "left")

    // 声明一个隐式的配置对象
    implicit val esConfig = ESConfig(config("es.httpHosts"), config("es.transportHosts"), config("es.index"), config("es.cluster.name"))
    // 将数据保存到 ES 中
    storeDataInES(movieWithTagsDF)

    // 关闭 SparkSession
    spark.stop()
  }
}

  def storeDataInMongDB(movieDF: DataFrame, ratingDF: DataFrame, tagDF: DataFrame)(implicit mongoConfig: MongoConfig): Unit = {
    // 新建一个到 MongoDB 的连接
    val mongoClient = MongoClient(MongoClientURI(mongoConfig.uri))
    // 如果 MongoDB 中已有相应的数据库，则先删除
    mongoClient(mongoConfig.db)(MONGODB_MOVIE_COLLECTION).dropCollection()
    mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION).dropCollection()
    mongoClient(mongoConfig.db)(MONGODB_TAG_COLLECTION).dropCollection()

    // 将 DF 数据写入对应的 MongoDB 表中
    movieDF.write
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_MOVIE_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()

    tagDF.write
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_TAG_COLLECTION)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

    // 对 MongoDB 中的数据表建索引
    mongoClient(mongoConfig.db)(MONGODB_MOVIE_COLLECTION).createIndex(MongoDBObject("mid" -> 1))
    mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION).createIndex(MongoDBObject("uid" -> 1))
    mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION).createIndex(MongoDBObject("mid" -> 1))
    mongoClient(mongoConfig.db)(MONGODB_TAG_COLLECTION).createIndex(MongoDBObject("uid" -> 1))
    mongoClient(mongoConfig.db)(MONGODB_TAG_COLLECTION).createIndex(MongoDBObject("mid" -> 1))

    // 关闭 MongoDB 的连接
    mongoClient.close()
  }

  def storeDataInES(movieWithTagsDF: DataFrame)(implicit esConfig: ESConfig): Unit = {
    // 新建一个 es 的配置
    val settings: Settings = Settings.builder().put("cluster.name", esConfig.clusterName).build()
    // 新建一个 es 的客户端
    val esClient = new PreBuiltTransportClient(settings)
    // 需要将 TransportHosts 添加到 esClient 中
    val REGEX_HOST_PORT = "(.+):(\\d+)".r
    esConfig.transportHosts.split(",").foreach {
      case REGEX_HOST_PORT(host: String, port: String) => {
        esClient.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), port.toInt))
      }
    }

    // 需要先清除掉 ES 中遗留的数据
    if (esClient.admin().indices().exists(new IndicesExistsRequest(esConfig.index))
      .actionGet()
      .isExists
    ) {
      esClient.admin().indices().delete(new DeleteIndexRequest(esConfig.index))
    }

    esClient.admin().indices().create(new CreateIndexRequest(esConfig.index))

    // 将数据写入到 ES 中
    movieWithTagsDF.write
      .option("es.nodes", esConfig.httpHosts)
      .option("es.http.timeout", "100m")
      .option("es.mapping.id", "mid") // 映射主键
      .mode("overwrite")
      .format("org.elasticsearch.spark.sql")
      .save(esConfig.index + "/" + ES_MOVIE_INDEX)
  }

    <dependencies>
        <!-- Spark 的依赖引入 -->
        <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>
        <!-- 引入 Scala -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
        </dependency>
        <!-- 加入 MongoDB 的驱动 -->
        <!-- 用于代码方式连接 MongoDB -->
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>casbah-core_2.11</artifactId>
            <version>${casbah.version}</version>
        </dependency>
        <!-- 用于 Spark 和 MongoDB 的对接 -->
        <dependency>
            <groupId>org.mongodb.spark</groupId>
            <artifactId>mongo-spark-connector_2.11</artifactId>
            <version>${mongodb-spark.version}</version>
        </dependency>
    </dependencies>

package com.atguigu.statistics

import java.text.SimpleDateFormat
import java.util.Date

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

case class Movie(mid: Int, name: String, descri: String, timelong: String, issue: String,
                 shoot: String, language: String, genres: String, actors: String, directors: String)

case class Rating(uid: Int, mid: Int, score: Double, timestamp: Long)

// 把 MongoDB 和 Elasticsearch 的配置封装成样例类
/**
  * @param uri MongDB 的连接
  * @param db  MongDB 的 数据库
  */
case class MongoConfig(uri: String, db: String)

case class Recommendation(mid: Int, score: Double)

// 定义电影类别 top10 推荐对象（每种类型的电影集合中评分最高的 10 个电影）
case class GenresRecommendation(genres: String, recs: Seq[Recommendation])

object StatisticsRecommender {
  // 定义 MongoDB 数据库中的一些表名
  val MONGODB_MOVIE_COLLECTION = "Movie"
  val MONGODB_RATING_COLLECTION = "Rating"

  val RATE_MORE_MOVIES = "RateMoreMovies" // 电影评分个数统计表
  val RATE_MORE_RECENTLY_MOVIES = "RateMoreRecentlyMovies" // 最近电影评分个数统计表
  val AVERAGE_MOVIES_Score = "AverageMoviesScore" // 电影平均评分表
  val GENRES_TOP_MOVIES = "GenresTopMovies" // 电影类别 TOP10

  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().setMaster(config("spark.cores")).setAppName("StatisticsRecommender")

    // 创建一个 SparkSession 对象
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    // 在对 DataFrame 和 Dataset 进行许多操作都需要这个包进行支持
    import spark.implicits._

    // 声明一个隐式的配置对象
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    // 从 MongoDB 中加载数据
    val movieDF = spark.read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_MOVIE_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[Movie] // DataSet
      .toDF()

    // 从 MongoDB 中加载数据
    val ratingDF = spark.read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_RATING_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[Rating] // DataSet
      .toDF()

    // 创建临时表，名为 ratings
    ratingDF.createOrReplaceTempView("ratings")

    // TODO:不同的统计推荐结果
    // ......

    // 关闭 SparkSession
    spark.stop()
  }

  def storeDFInMongDB(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()
  }
}

    // 1、历史热门电影统计：根据所有历史评分数据，计算历史评分次数最多的电影。mid,count
    val rateMoreMoviesDF = spark.sql("select mid, count(mid) as count from ratings group by mid")
    // 把结果写入对应的 MongoDB 表中
    storeDFInMongDB(rateMoreMoviesDF, RATE_MORE_MOVIES)

    // 2、最近热门电影统计：根据评分次数，按月为单位计算最近时间的月份里面评分次数最多的电影集合。mid,count,yearmonth
    // 创建一个日期格式化工具
    val simpleDateFormat = new SimpleDateFormat("yyyyMM")
    // 注册一个 UDF 函数，用于将 timestamp 转换成年月格式    1260759144000 => 201605
    spark.udf.register("changeDate", (x: Long) => {
      simpleDateFormat.format(new Date(x * 1000)).toInt
    })
    // 对原始数据 ratings 做预处理，去掉 uid，保存成临时表，名为 ratingOfMonth
    val ratingOfYearMonth = spark.sql("select mid, score, changeDate(timestamp) as yearmonth from ratings")
    ratingOfYearMonth.createOrReplaceTempView("ratingOfMonth")
    // 根据评分次数，按月为单位计算最近时间的月份里面评分次数最多的电影集合
    val rateMoreRecentlyMoviesDF = spark.sql("select mid, count(mid), yearmonth as count from ratingOfMonth group by yearmonth, mid order by yearmonth desc, count desc")
    // 把结果写入对应的 MongoDB 表中
    storeDFInMongDB(rateMoreRecentlyMoviesDF, RATE_MORE_RECENTLY_MOVIES)

    // 3、电影平均得分统计：根据历史数据中所有用户对电影的评分，周期性的计算每个电影的平均得分。mid,avg
    val averageMoviesDF = spark.sql("select mid, avg(score) as avg from ratings group by mid")
    // 把结果写入对应的 MongoDB 表中
    storeDFInMongDB(averageMoviesDF, AVERAGE_MOVIES_Score)

    // 4、每个类别优质电影统计：根据提供的所有电影类别，分别计算每种类型的电影集合中评分最高的 10 个电影。
    // 定义所有的类别
    val genres = List("Action", "Adventure", "Animation", "Comedy", "Crime", "Documentary", "Drama", "Famil y", "Fantasy",
      "Foreign", "History", "Horror", "Music", "Mystery", "Romance", "Science", "Tv", "Thriller", "War", "Western")
    // 把电影的平均评分加入到 movie 表中，使用 inner join，不满足条件的不显示
    val movieWithScore = movieDF.join(averageMoviesDF, "mid")
    // 为做笛卡尔积，我们需要把 genres 转成 RDD
    val genresRDD = spark.sparkContext.makeRDD(genres)
    // 计算类别 top10，首先对类别和电影做笛卡尔积，然后进行过滤
    val genresTopMoviesDF = genresRDD.cartesian(movieWithScore.rdd)
      .filter {
        // 条件过滤：找出 movie 中的字段 genres 值包含当前类别 genre 的那些
        case (genre, movieRow) => movieRow.getAs[String]("genres").toLowerCase().contains(genre.toLowerCase())
      }
      .map { // 将整个数据集的数据量减小，生成 RDD[String, Iter[mid, avg]]
        case (genre, movieRow) => (genre, (movieRow.getAs[Int]("mid"), movieRow.getAs[Double]("avg")))
      }
      .groupByKey()
      .map {
        case (genre, items) => GenresRecommendation(genre, items.toList.sortWith(_._2 > _._2).take(10).map(item =>
          Recommendation(item._1, item._2)
        ))
      }
      .toDF()
    // 把结果写入对应的 MongoDB 表中
    storeDFInMongDB(genresTopMoviesDF, GENRES_TOP_MOVIES)

    <dependencies>
        <!-- Java 中线性代数相关的库 -->
        <dependency>
            <groupId>org.scalanlp</groupId>
            <artifactId>jblas</artifactId>
            <version>${jblas.version}</version>
        </dependency>
        <!-- Spark 的依赖引入 -->
        <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.apache.spark</groupId>
            <artifactId>spark-mllib_2.11</artifactId>
        </dependency>
        <!-- 引入 Scala -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
        </dependency>
        <!-- 加入 MongoDB 的驱动 -->
        <!-- 用于代码方式连接 MongoDB -->
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>casbah-core_2.11</artifactId>
            <version>${casbah.version}</version>
        </dependency>
        <!-- 用于 Spark 和 MongoDB 的对接 -->
        <dependency>
            <groupId>org.mongodb.spark</groupId>
            <artifactId>mongo-spark-connector_2.11</artifactId>
            <version>${mongodb-spark.version}</version>
        </dependency>
    </dependencies>

package com.atguigu.offline

import org.apache.spark.SparkConf
import org.apache.spark.mllib.recommendation.{ALS, Rating}
import org.apache.spark.sql.SparkSession
import org.jblas.DoubleMatrix

// 基于评分数据的 LFM，只需要 rating 数据（用户评分表）注意：spark mllib 中有 Rating 类，为了便于区别，我们重新命名为 MovieRating
case class MovieRating(uid: Int, mid: Int, score: Double, timestamp: Long)

case class MongoConfig(uri: String, db: String)

// 标准推荐对象
case class Recommendation(mid: Int, score: Double)

// 用户推荐列表
case class UserRecs(uid: Int, recs: Seq[Recommendation])

// 电影相似度（电影推荐）
case class MovieRecs(mid: Int, recs: Seq[Recommendation])

object OfflineRecommender {
  // 定义 MongoDB 数据库中的一些表名
  val MONGODB_RATING_COLLECTION = "Rating"

  // 推荐表的名称
  val USER_RECS = "UserRecs"
  val MOVIE_RECS = "MovieRecs"

  val USER_MAX_RECOMMENDATION = 20

  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().setMaster(config("spark.cores")).setAppName("OfflineRecommender")

    // 创建一个 SparkSession 对象
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    // 在对 DataFrame 和 Dataset 进行许多操作都需要这个包进行支持
    import spark.implicits._

    // 声明一个隐式的配置对象
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    // 从 MongoDB 中加载数据
    val ratingRDD = spark.read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_RATING_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[MovieRating] // DataSet
      .rdd
      .map(rating => (rating.uid, rating.mid, rating.score)) // 转换成 RDD，并且去掉时间戳
      .cache()

    // 从 ratingRDD 数据中提取所有的 uid 和 mid ，并去重
    val userRDD = ratingRDD.map(_._1).distinct()
    val movieRDD = ratingRDD.map(_._2).distinct()

    // 训练隐语义模型
    val trainData = ratingRDD.map(x => Rating(x._1, x._2, x._3))

    val (rank, iterations, lambda) = (50, 5, 0.01)
    val model = ALS.train(trainData, rank, iterations, lambda)

    // 基于用户和电影的隐特征，计算预测评分，得到用户推荐列表
    // user 和 movie 做笛卡尔积，得到一个空评分矩阵，即产生 (uid，mid) 的元组
    val userMovies = userRDD.cartesian(movieRDD)

    // 调用 model 的 predict 方法进行预测评分
    val preRatings = model.predict(userMovies)

    val userRecs = preRatings
      .filter(_.rating > 0) // 过滤出评分大于零的项
      .map(rating => (rating.user, (rating.product, rating.rating)))
      .groupByKey()
      .map {
        case (uid, recs) => UserRecs(uid, recs.toList.sortWith(_._2 > _._2).take(USER_MAX_RECOMMENDATION).map(x => Recommendation(x._1, x._2)))
      }
      .toDF()

    // 把结果写入对应的 MongoDB 表中
    userRecs.write
      .option("uri", mongoConfig.uri)
      .option("collection", USER_RECS)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

    // TODO:计算电影相似度矩阵

    spark.stop()
}

    // 基于电影的隐特征，计算相似度矩阵，得到电影的相似度列表
    val movieFeatures = model.productFeatures.map {
      case (mid, features) => (mid, new DoubleMatrix(features))
    }

    // 对所有电影两两计算它们的相似度，先做笛卡尔积
    val movieRecs = movieFeatures.cartesian(movieFeatures)
      .filter {
        // 把自己跟自己的配对过滤掉
        case (a, b) => a._1 != b._1
      }
      .map {
        case (a, b) => {
          val simScore = this.consinSim(a._2, b._2)
          (a._1, (b._1, simScore))
        }
      }
      .filter(_._2._2 > 0.6) // 过滤出相似度大于 0.6 的
      .groupByKey()
      .map {
        case (mid, recs) => MovieRecs(mid, recs.toList.sortWith(_._2 > _._2).map(x => Recommendation(x._1, x._2)))
      }
      .toDF()

    // 把结果写入对应的 MongoDB 表中
    movieRecs.write
      .option("uri", mongoConfig.uri)
      .option("collection", MOVIE_RECS)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

  // 求两个向量的余弦相似度
  def consinSim(movie1: DoubleMatrix, movie2: DoubleMatrix): Double = {
    movie1.dot(movie2) / (movie1.norm2() * movie2.norm2()) // l1范数：向量元素绝对值之和；l2范数：即向量的模长（向量的长度）,向量元素的平方和再开方
  }

package com.atguigu.offline

import breeze.numerics.sqrt
import com.atguigu.offline.OfflineRecommender.MONGODB_RATING_COLLECTION
import org.apache.spark.SparkConf
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object ALSTrainer {

  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().setMaster(config("spark.cores")).setAppName("OfflineRecommender")

    // 创建一个 SparkSession 对象
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    // 在对 DataFrame 和 Dataset 进行许多操作都需要这个包进行支持
    import spark.implicits._

    // 声明一个隐式的配置对象
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    // 从 MongoDB 中加载数据
    val ratingRDD = spark.read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_RATING_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[MovieRating] // DataSet
      .rdd
      .map(rating => Rating(rating.uid, rating.mid, rating.score)) // 转换成 RDD，并且去掉时间戳
      .cache()

    // 将一个 RDD 随机切分成两个 RDD，用以划分训练集和测试集
    val splits = ratingRDD.randomSplit(Array(0.8, 0.2))
    val trainingRDD = splits(0)
    val testingRDD = splits(1)

    // 模型参数选择，输出最优参数
    adjustALSParam(trainingRDD, testingRDD)

    spark.close()
  }

  def adjustALSParam(trainData: RDD[Rating], testData: RDD[Rating]): Unit = {
    // 这里指定迭代次数为 5，rank 和 lambda 在几个值中选取调整
    val result = for (rank <- Array(50, 100, 200, 300); lambda <- Array(1, 0.1, 0.01, 0.001))
      yield {
        val model = ALS.train(trainData, rank, 5, lambda)
        val rmse = getRMSE(model, testData)
        (rank, lambda, rmse)
      }
    // 控制台打印输出
    // println(result.sortBy(_._3).head)
    println(result.minBy(_._3))
  }

  def getRMSE(model: MatrixFactorizationModel, data: RDD[Rating]): Double = {
    // 计算预测评分
    val userProducts = data.map(item => (item.user, item.product))
    val predictRating = model.predict(userProducts)

    // 以 uid,mid 作为外键，将 实际观测值 和 预测值 使用内连接
    val observed = data.map(item => ((item.user, item.product), item.rating))
    val predicted = predictRating.map(item => ((item.user, item.product), item.rating))
    // 内连接，得到 (uid, mid), (observe, predict)

    // 计算 RMSE
    sqrt(
      observed.join(predicted).map {
        case ((uid, mid), (observe, predict)) =>
          val err = observe - predict
          err * err
      }.mean()
    )
  }

    <dependencies>
        <!-- Spark 的依赖引入 -->
        <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.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
        </dependency>
        <!-- 引入 Scala -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
        </dependency>
        <!-- 加入 MongoDB 的驱动 -->
        <!-- 用于代码方式连接 MongoDB -->
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>casbah-core_2.11</artifactId>
            <version>${casbah.version}</version>
        </dependency>
        <!-- 用于 Spark 和 MongoDB 的对接 -->
        <dependency>
            <groupId>org.mongodb.spark</groupId>
            <artifactId>mongo-spark-connector_2.11</artifactId>
            <version>${mongodb-spark.version}</version>
        </dependency>
        <!-- redis -->
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.0</version>
        </dependency>
        <!-- kafka -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.2.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
    </dependencies>

package com.atguigu.streaming

import com.mongodb.casbah.commons.MongoDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

// 定义连接助手对象并序列化
object ConnHelper extends Serializable {
  lazy val jedis = new Jedis("hadoop102")
  lazy val mongoClient = MongoClient(MongoClientURI("mongodb://localhost:27017/recommender"))
}

case class MongoConfig(uri: String, db: String)

// 定义一个基准推荐对象
case class Recommendation(mid: Int, score: Double)

// 定义基于预测评分的用户推荐列表
case class UserRecs(uid: Int, recs: Seq[Recommendation])

// 定义基于LFM电影特征向量的电影相似度列表
case class MovieRecs(mid: Int, recs: Seq[Recommendation])

object StreamingRecommender {

  val MONGODB_STREAM_RECS_COLLECTION = "StreamRecs"
  val MONGODB_RATING_COLLECTION = "Rating"
  val MONGODB_MOVIE_RECS_COLLECTION = "MovieRecs"

  val MAX_USER_RATINGS_NUM = 20
  val MAX_SIM_MOVIES_NUM = 20

  def main(args: Array[String]): Unit = {

  }
}

  def main(args: Array[String]): Unit = {
    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://localhost:27017/recommender",
      "mongo.db" -> "recommender",
      "kafka.topic" -> "recommender"
    )

    // 创建一个 SparkConf 对象
    val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("StreamingRecommender").set("spark.ui.port", "44040" )

    // 创建一个 SparkSession 对象
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    // 获取 Streaming Context
    val sc = spark.sparkContext
    val ssc = new StreamingContext(sc, Seconds(2)) // 微批次处理时间间隔

    // 在对 DataFrame 和 Dataset 进行许多操作都需要这个包进行支持
    import spark.implicits._

    // 声明一个隐式的配置对象
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    // 加载数据：电影相似度矩阵数据，转换成为 Map[Int, Map[Int, Double]]，把它广播出去
    val simMovieMatrix = spark.read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_MOVIE_RECS_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[MovieRecs]
      .rdd
      .map { movieRecs => // 为了查询相似度方便，转换成 KV
        (movieRecs.mid, movieRecs.recs.map(x => (x.mid, x.score)).toMap)
      }.collectAsMap()

    val simMovieMatrixBroadCast = sc.broadcast(simMovieMatrix)

    // 定义 Kafka 的连接参数
    val kafkaParam = Map(
      "bootstrap.servers" -> "hadoop102:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "recommender",
      "auto.offset.reset" -> "latest" // 偏移量的初始设置
    )

    // 通过 Kafka 创建一个 DStream
    val kafkaStream = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Array(config("kafka.topic")), kafkaParam)
    )

    // 把原始数据 uid|mid|score|timestamp 转换成评分流
    val ratingStream = kafkaStream.map {
      msg =>
        val attr = msg.value().split("\\|")
        (attr(0).toInt, attr(1).toInt, attr(2).toDouble, attr(3).toLong)
    }

    // 继续做流式处理，实时算法部分
    ratingStream.foreachRDD {
      rdds =>
        rdds.foreach {
          case (uid, mid, score, timestamp) => {
            println("rating data coming! >>>>>>>>>>>>>>>>>>>>")
            // 1、从 redis 中获取当前用户最近的 K 次评分，保存成 Array[(mid, score)]
            val userRecentlyRatings = getUserRecentlyRatings(MAX_USER_RATINGS_NUM, uid, ConnHelper.jedis)

            // 2、从相思度矩阵中取出当前电影最相似的 N 个电影，作为备选电影列表，Array[mid]
            val candidateMovies = getTopsSimMovies(MAX_SIM_MOVIES_NUM, mid, uid, simMovieMatrixBroadCast.value)

            // 3、对每个备选电影，计算推荐优先级，得到当前用户的实时推荐列表，Array[(mid, score)]
            val streamRecs = computeMovieScores(candidateMovies, userRecentlyRatings, simMovieMatrixBroadCast.value)

            // 4、把推荐数据保存到 MongoDB 中
            storeDataInMongDB(uid, streamRecs)
          }
        }
    }

    // 开始接收和处理数据
    ssc.start()

    println(">>>>>>>>>>>>>>>>>>>> streaming started!")

    ssc.awaitTermination()
  }

  // 因为 redis 操作返回的是 java 类，为了使用 map 操作需要引入转换类
  import scala.collection.JavaConversions._

  /**
    * 获取当前最近的 K 次电影评分
    *
    * @param num 评分的个数
    * @param uid 谁的评分
    * @return
    */
  def getUserRecentlyRatings(num: Int, uid: Int, jedis: Jedis): Array[(Int, Double)] = {
    // 从 redis 中读取数据，用户评分数据保存在 uid:UID 为 key 的队列中，里面的 value 是 MID:SCORE
    jedis.lrange("uid:" + uid.toString, 0, num) // 从用户的队列中取出 num 个评分
      .map {
      item => // 具体的每一个评分是以冒号分割的两个值
        val attr = item.split("\\:")
        (attr(0).trim.toInt, attr(1).trim.toDouble)
    }
      .toArray
  }

  /**
    * 获取与当前电影 K 个相似的电影，作为备选电影
    *
    * @param num         相似电影的数量
    * @param mid         当前电影的 ID
    * @param uid         当前的评分用户 ID
    * @param simMovies   电影相似度矩阵的广播变量值
    * @param mongoConfig MongoDB 的配置
    * @return 过滤之后的备选电影列表
    */
  def getTopsSimMovies(num: Int, mid: Int, uid: Int, simMovies: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
                      (implicit mongoConfig: MongoConfig): Array[Int] = {
    // 1、从相似度矩阵中拿到所有相似的电影
    val allSimMovies = simMovies(mid).toArray

    // 2、从 MongnDB 中查询用户已经看过的电影
    val ratingExist = ConnHelper.mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION)
      .find(MongoDBObject("uid" -> uid))
      .toArray
      .map {
        item => item.get("mid").toString.toInt
      }

    // 3、把看过的过滤掉，得到备选电影的输出列表
    allSimMovies.filter(x => !ratingExist.contains(x._1))
      .sortWith(_._2 > _._2)
      .take(num)
      .map(x => x._1)
  }

  /**
    * 计算待选电影的推荐分数
    *
    * @param candidateMovies     与当前电影最相似的 N 个电影（待选电影）
    * @param userRecentlyRatings 用户最近的 K 次评分
    * @param simMovies           电影相似度矩阵的广播变量值
    * @return
    */
  def computeMovieScores(candidateMovies: Array[Int], userRecentlyRatings: Array[(Int, Double)],
                         simMovies: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Array[(Int, Double)] = {
    // 定义一个 ArrayBuffer，用于保存每一个备选电影的基础得分
    val scores = scala.collection.mutable.ArrayBuffer[(Int, Double)]()
    // 定义一个 HashMap，保存每一个备选电影的增强减弱因子
    val increMap = scala.collection.mutable.HashMap[Int, Int]()
    val decreMap = scala.collection.mutable.HashMap[Int, Int]()

    for (candidateMovies <- candidateMovies; userRecentlyRatings <- userRecentlyRatings) {

      // 获取备选电影和最近评分电影的相似度的得分
      val simScore = getMoviesSimScore(candidateMovies, userRecentlyRatings._1, simMovies)

      if (simScore > 0.7) {
        // 计算候选电影的基础推荐得分
        scores += ((candidateMovies, simScore * userRecentlyRatings._2))
        if (userRecentlyRatings._2 > 3) {
          increMap(candidateMovies) = increMap.getOrDefault(candidateMovies, 0) + 1
        } else {
          decreMap(candidateMovies) = decreMap.getOrDefault(candidateMovies, 0) + 1
        }
      }
    }

    // 根据备选电影的 mid 做 groupBy，根据公式求最后的推荐得分
    scores.groupBy(_._1).map {
      // groupBy 之后得到的数据是 Map(mid -> ArrayBuffer[(mid, score)])
      case (mid, scoreList) =>
        (mid, scoreList.map(_._2).sum / scoreList.length + log(increMap.getOrDefault(mid, 1)) - log(decreMap.getOrDefault(mid, 1)))
    }.toArray
  }

  /**
    * 获取备选电影和最近评分电影的相似度的得分
    *
    * @param mid1      备选电影
    * @param mid2      最近评分电影
    * @param simMovies 电影相似度矩阵的广播变量值
    * @return
    */
  def getMoviesSimScore(mid1: Int, mid2: Int,
                        simMovies: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Double = {
    simMovies.get(mid1) match {
      case Some(sims) => sims.get(mid2) match {
        case Some(score) => score
        case None => 0.0
      }
      case None => 0.0
    }
  }

  /**
    * 求一个数的对数，底数默认为 10
    *
    * @param m
    * @return
    */
  def log(m: Int): Double = {
    val N = 10
    math.log(m) / math.log(N)
  }

  /**
    * 把结果写入对应的 MongoDB 表中
    *
    * @param uid
    * @param streamRecs 流式的推荐结果
    * @param mongoConfig MongoDB 的配置
    */
  def storeDataInMongDB(uid: Int, streamRecs: Array[(Int, Double)])(implicit mongoConfig: MongoConfig): Unit = {
    // 定义到 MongoDB 中 StreamRecs 表的连接
    val streamRecsCollection = ConnHelper.mongoClient(mongoConfig.db)(MONGODB_STREAM_RECS_COLLECTION)
    // 如果表中已有 uid 对应的数据，则删除
    streamRecsCollection.findAndRemove(MongoDBObject("uid" -> uid))
    // 将新的 streamRecs 存入表 StreamRecs 中
    streamRecsCollection.insert(MongoDBObject("uid" -> uid, "recs" -> streamRecs.map(x => MongoDBObject("mid" -> x._1, "score" -> x._2))))
  }

C:\Windows\system32>mongod

[atguigu@hadoop102 bin]$ pwd
/usr/local/bin
[atguigu@hadoop102 bin]$ ./redis-server /opt/module/redis-3.0.4/myredis/redis.conf

[atguigu@hadoop102 bin]$ ./redis-cli -p 6379

[atguigu@hadoop102 zookeeper-3.4.10]$ bin/zkServer.sh start
[atguigu@hadoop103 zookeeper-3.4.10]$ bin/zkServer.sh start
[atguigu@hadoop104 zookeeper-3.4.10]$ bin/zkServer.sh start

[atguigu@hadoop102 kafka]$ bin/kafka-server-start.sh config/server.properties &
[atguigu@hadoop103 kafka]$ bin/kafka-server-start.sh config/server.properties &
[atguigu@hadoop104 kafka]$ bin/kafka-server-start.sh config/server.properties &

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>recommender</artifactId>
        <groupId>com.atguigu</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>KafkaStreaming</artifactId>

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
            <version>0.10.2.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.2.1</version>
        </dependency>
    </dependencies>

    <build>
        <finalName>kafkastream</finalName>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>com.atguigu.kafkastream.Application</mainClass>
                        </manifest>
                    </archive>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

package com.atguigu.kafkastream;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.TopologyBuilder;

import java.util.Properties;

public class Application {
    public static void main(String[] args) {
        String brokers = "hadoop102:9092";
        String zookeepers = "hadoop102:2181";

        // 输入和输出的 topic
        String from = "log";
        String to = "recommender";

        // 定义 kafka streaming 的配置
        Properties settings = new Properties();
        settings.put(StreamsConfig.APPLICATION_ID_CONFIG, "logFilter");
        settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        settings.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, zookeepers);

        // 创建 kafka streaming 配置对象
        StreamsConfig config = new StreamsConfig(settings);

        // 创建一个拓扑建构器
        TopologyBuilder builder = new TopologyBuilder();

        // 定义流处理的拓扑结构
        builder.addSource("SOURCE", from)
                .addProcessor("PROCESSOR", () -> new LogProcessor(), "SOURCE")
                .addSink("SINK", to, "PROCESSOR");

        // 创建 KafkaStreams 对象
        KafkaStreams streams = new KafkaStreams(builder, config);

        streams.start();

        System.out.println("kafka stream started! >>>>>>>>>>>>>>>>>>>>");
    }
}

package com.atguigu.kafkastream;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;

public class LogProcessor implements Processor<byte[], byte[]> {

    private ProcessorContext context;

    @Override
    public void init(ProcessorContext processorContext) {
        this.context = processorContext;
    }

    @Override
    public void process(byte[] dummy, byte[] line) { // dummy 表示 哑变量，没什么用
        // 把收集到的日志信息用 String 表示
        String input = new String(line);
        // 根据前缀 MOVIE_RATING_PREFIX: 从日志信息中提取评分数据
        if (input.contains("MOVIE_RATING_PREFIX:")) {
            System.out.println("movie rating data coming! >>>>>>>>>>>>>>>>>>>>" + input);

            input = input.split("MOVIE_RATING_PREFIX:")[1].trim();
            context.forward("logProcessor".getBytes(), input.getBytes());
        }
    }

    @Override
    public void punctuate(long l) {

    }

    @Override
    public void close() {

    }
}

agent.sources = exectail 
agent.channels = memoryChannel 
agent.sinks = kafkasink

# For each one of the sources, the type is defined.
agent.sources.exectail.type = exec

# 下面这个路径是需要收集日志的绝对路径，改为自己的日志目录
agent.sources.exectail.command = tail –f /opt/module/flume/log/a gent.log
agent.sources.exectail.interceptors = i1
agent.sources.exectail.interceptors.i1.type = regex_filter
# 定义日志过滤前缀的正则
agent.sources.exectail.interceptors.i1.regex = .+MOVIE_RATING_PREFIX.+

# The channel can be defined as follows.
agent.sources.exectail.channels = memoryChannel

# Each sink's type must be defined.
agent.sinks.kafkasink.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafkasink.kafka.topic = log
agent.sinks.kafkasink.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092
agent.sinks.kafkasink.kafka.producer.acks = 1
agent.sinks.kafkasink.kafka.flumeBatchSize = 20

# Specify the channel the sink should use.
agent.sinks.kafkasink.channel = memoryChannel
# Each channel's type is defined. 
agent.channels.memoryChannel.type = memory

# Other config values specific to each type of channel(sink or source)
# can be defined as well
# In this case, it specifies the capacity of the memory channel.
agent.channels.memoryChannel.capacity = 10000

[atguigu@hadoop102 flume]$ bin/flume-ng agent \
--conf conf/ --name a1 --conf-file job/flume-log-kafka.conf \
-Dflume.root.logger=INFO,console

package com.atguigu.content

import org.apache.spark.SparkConf
import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}
import org.apache.spark.ml.linalg.SparseVector
import org.apache.spark.sql.SparkSession
import org.jblas.DoubleMatrix

// 需要的数据源是电影内容信息
case class Movie(mid: Int, name: String, descri: String, timelong: String, issue: String,
                 shoot: String, language: String, genres: String, actors: String, directors: String)

case class MongoConfig(uri: String, db: String)

// 定义一个基准推荐对象
case class Recommendation(mid: Int, score: Double)

// 定义电影内容信息提取出的特征向量的电影相似度列表
case class MovieRecs(mid: Int, recs: Seq[Recommendation])

object ContentRecommender {

  // 定义表名和常量
  val MONGODB_MOVIE_COLLECTION = "Movie"
  val CONTENT_MOVIE_RECS = "ContentMovieRecs"

  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().setMaster(config("spark.cores")).setAppName("ContentRecommender")

    // 创建一个 SparkSession 对象
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    // 声明一个隐式的配置对象
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    // 在对 DataFrame 和 Dataset 进行许多操作都需要这个包进行支持
    import spark.implicits._

    // 加载数据，并作预处理
    val movieTagsDF = spark.read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_MOVIE_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[Movie]
      .map { // 提取 mid,name,genres 三项作为原始的内容特征，分词器默认分隔符是空格
        x => (x.mid, x.name, x.genres.map(c => if (c == '|') ' ' else c))
      }
      .toDF("mid", "name", "genres")
      .cache()

    // TODO:从内容信息中提取电影特征的特征向量
    // 创建一个分词器，默认按照空格分词
    val tokenizer = new Tokenizer().setInputCol("genres").setOutputCol("words")
    // 用分词器对原始数据进行转换，生成新的一列words
    val wordsData = tokenizer.transform(movieTagsDF)

    // 引入 HashingTF 工具，该工具可以将词语序列转换成对应的词频
    val hashingTF = new HashingTF().setInputCol("words").setOutputCol("rawFeatures").setNumFeatures(50)
    val featurizeData = hashingTF.transform(wordsData)

    // 测试
    // wordsData.show()
    // featurizeData.show()
    // featurizeData.show(truncate = false) // 不压缩显示

    // 引入 IDF 工具，该工具可以得到 IDF 模型
    val idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")
    // 训练 IDF 模型，得到每个词的逆文档频率
    val idfModel = idf.fit(featurizeData)

    // 用 IDF 模型对原数据进行处理，得到文档中每个词的 TF-IDF，作为新的特征向量
    val rescaleData = idfModel.transform(featurizeData)

    // 测试
    // rescaleData.show(truncate = false) // 不压缩显示

    val movieFeatures = rescaleData.map(
      row => (row.getAs[Int]("mid"), row.getAs[SparseVector]("features").toArray)
    ).rdd.map(
      x => (x._1, new DoubleMatrix(x._2))
    )

    // 测试
    // movieFeatures.collect().foreach(println)

    // 对所有电影两两计算它们的相似度，先做笛卡尔积
    val movieRecs = movieFeatures.cartesian(movieFeatures)
      .filter {
        // 把自己跟自己的配对过滤掉
        case (a, b) => a._1 != b._1
      }
      .map {
        case (a, b) => {
          val simScore = this.consinSim(a._2, b._2)
          (a._1, (b._1, simScore))
        }
      }
      .filter(_._2._2 > 0.6) // 过滤出相似度大于 0.6 的
      .groupByKey()
      .map {
        case (mid, recs) => MovieRecs(mid, recs.toList.sortWith(_._2 > _._2).map(x => Recommendation(x._1, x._2)))
      }
      .toDF()

    // 把结果写入对应的 MongoDB 表中
    movieRecs.write
      .option("uri", mongoConfig.uri)
      .option("collection", CONTENT_MOVIE_RECS)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

    spark.stop()
  }

  // 求两个向量的余弦相似度
  def consinSim(movie1: DoubleMatrix, movie2: DoubleMatrix): Double = {
    movie1.dot(movie2) / (movie1.norm2() * movie2.norm2()) // l1范数：向量元素绝对值之和；l2范数：即向量的模长（向量的长度）,向量元素的平方和再开方
  }
}