从零开发，如何基于 Spark 构建一个电影推荐系统

create table movies(
	id int(8) not null primary key,
	title varchar(200) not null,
	genres text
);

create table tags(
	userId varchar(8) default '',
	movieId varchar(8) default '',
	tag text,
	timestamp varchar(16) default ''
);

LOAD DATA INFILE '/var/lib/mysql-files/ml-32m/movies.csv'
INTO TABLE movies
FIELDS TERMINATED BY ',' 
ENCLOSED BY '"'
LINES TERMINATED BY '\n'
IGNORE 1 ROWS
(id, title, genres);

LOAD DATA INFILE '/var/lib/mysql-files/ml-32m/tags.csv'
INTO TABLE tags
FIELDS TERMINATED BY ',' 
ENCLOSED BY '"'
LINES TERMINATED BY '\n'
IGNORE 1 ROWS
(userId, movieId, tag, timestamp);

case class Rating(userId: Int, movieId: Int, rating: Float, timestamp: Long)
def parseRating(str: String): Rating = {
  val fields = str.split(",")
  assert(fields.size == 4)
  Rating(fields(0).toInt, fields(1).toInt, fields(2).toFloat, fields(3).toLong)
  }

def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder
      .appName("ALSExample")
      .master("local[4]")
      .getOrCreate()

    import spark.implicits._

    // 读取评分数据
    val ratings = spark.read.textFile("path/ml-32m/ratings.csv")
      .map(parseRating)
      .toDF()

    // 划分训练集和测试集
    val Array(training, test) = ratings.randomSplit(Array(0.8, 0.2))

    // 构建 ALS 模型
    val als = new ALS()
      .setMaxIter(5)
      .setRegParam(0.01)
      .setUserCol("userId")
      .setItemCol("movieId")
      .setRatingCol("rating")

    // 模型训练
    val model = als.fit(training)

    // 模型评估
    model.setColdStartStrategy("drop")
    val predictions = model.transform(test)

    val evaluator = new RegressionEvaluator()
      .setMetricName("rmse")
      .setLabelCol("rating")
      .setPredictionCol("prediction")

    val rmse = evaluator.evaluate(predictions)
    println(s"Root-mean-square error = $rmse")

    // 获取每个用户 top10 推荐
    val userRecs = model.recommendForAllUsers(10)
}

CREATE TABLE user_movie_recommendations (
  user_id INT NOT NULL,
  movie_id INT NOT NULL,
  prediction FLOAT,
  recommend_time DATETIME DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (user_id, movie_id)
);

val explodedRecs = userRecs
  .withColumn("recommendation", F.explode(F.col("recommendations")))
  .select(
    F.col("userId").as("user_id"),
    F.col("recommendation.movieId").as("movie_id"),
    F.col("recommendation.rating").as("prediction"),
    F.current_timestamp().as("recommend_time")
  )

// MySQL JDBC 配置
val mysqlUrl = "jdbc:mysql://127.0.0.1:3306/movies"
val mysqlProperties = new java.util.Properties()
mysqlProperties.setProperty("user", "root")
mysqlProperties.setProperty("password", "123456")
mysqlProperties.setProperty("driver", "com.mysql.cj.jdbc.Driver")

// 写入 MySQL，14 分区并发写，每批 2000 条，提升性能
explodedRecs
  .repartition(14)
  .write
  .mode("append")
  .option("batchsize", "2000")
  .option("isolationLevel", "NONE")
  .jdbc(mysqlUrl, "user_movie_recommendations", mysqlProperties)

spark.stop()

// 根据ID查询电影
@Select("SELECT * FROM movie WHERE id = #{id}")
Movie findById(Long id);

// 根据类型查询电影
@Select("SELECT * FROM movie WHERE genre = #{genre} AND status = 1")
List<Movie> findByGenre(String genre);

// 获取电影列表
@PostMapping("/list")
public Result getMovieList(@RequestHeader("Authorization") String token) {
    try {
        if (token != null && token.startsWith("Bearer ")) {
            token = token.substring(7);
            if (!JwtUtil.validateToken(token)) {
                return Result.error("Invalid token");
            }
        }

        logger.info("User getting movie list");
        List<Movie> movies = movieService.findAll();
        return Result.success(movies);
    } catch (Exception e) {
        logger.error("Failed to get movie list: {}", e.getMessage());
        return Result.error(e.getMessage());
    }
}

// 获取电影详情
@PostMapping("/detail")
public Result getMovieDetail(@RequestHeader("Authorization") String token, @RequestBody Map<String, Long> request) {
    try {
        Long movieId = request.get("id");
        logger.info("User getting movie detail for id: {}", movieId);
        Movie movie = movieService.findById(movieId);
        if (movie != null) {
            return Result.success(movie);
        } else {
            return Result.error("Movie not found");
        }
    } catch (Exception e) {
        return Result.error(e.getMessage());
    }
}

// 添加评分
@PostMapping("/add")
public Result addRating(@RequestHeader("Authorization") String token, @RequestBody Map<String, Object> request) {
    try {
        if (token != null && token.startsWith("Bearer ")) {
            token = token.substring(7);
            if (!JwtUtil.validateToken(token)) {
                return Result.error("Invalid token");
            }
            Long userId = JwtUtil.getUserIdFromToken(token);
            Long movieId = Long.valueOf(request.get("movieId").toString());
            Integer rating = Integer.valueOf(request.get("rating").toString());
            String comment = (String) request.get("comment");
            logger.info("User {} adding rating {} for movie {} with comment: {}", userId, rating, movieId, comment);
            Rating savedRating = ratingService.addRating(userId, movieId, rating, comment);
            return Result.success(savedRating);
        }
        return Result.error("Invalid token");
    } catch (Exception e) {
        logger.error("Failed to add rating: {}", e.getMessage(), e);
        return Result.error(e.getMessage());
    }
}

/ 插入新评分
@Insert("INSERT INTO rating(user_id, movie_id, rating, comment, create_time, update_time) " +
        "VALUES(#{userId}, #{movieId}, #{rating}, #{comment}, #{createTime}, #{updateTime})")
@Options(useGeneratedKeys = true, keyProperty = "id")
int insert(Rating rating);

// 根据ID查询评分（包含用户名和电影标题）
@Select("SELECT r.*, u.username, m.title as movieTitle FROM rating r " +
        "LEFT JOIN user u ON r.user_id = u.id " +
        "LEFT JOIN movie m ON r.movie_id = m.id " +
        "WHERE r.id = #{id}")
Rating findById(Long id);

// 查询用户的所有评分
@Select("SELECT r.*, u.username, m.title as movieTitle FROM rating r " +
        "LEFT JOIN user u ON r.user_id = u.id " +
        "LEFT JOIN movie m ON r.movie_id = m.id " +
        "WHERE r.user_id = #{userId} ORDER BY r.create_time DESC")
List<Rating> findByUserId(Long userId);

// 查询电影的所有评分
@Select("SELECT r.*, u.username, m.title as movieTitle FROM rating r " +
        "LEFT JOIN user u ON r.user_id = u.id " +
        "LEFT JOIN movie m ON r.movie_id = m.id " +
        "WHERE r.movie_id = #{movieId} ORDER BY r.create_time DESC")
List<Rating> findByMovieId(Long movieId);

// 查询用户的所有推荐
@Select("SELECT r.*, u.username, m.title as movie_title, m.poster_url FROM recommendation r " +
        "LEFT JOIN user u ON r.user_id = u.id " +
        "LEFT JOIN movie m ON r.movie_id = m.id " +
        "WHERE r.user_id = #{userId} ORDER BY r.score DESC")
List<Recommendation> findByUserId(Long userId);

// 获取用户的推荐列表
@PostMapping("/my-list")
public Result getMyRecommendationList(@RequestHeader("Authorization") String token) {
    try {
        if (token != null && token.startsWith("Bearer ")) {
            token = token.substring(7);
            if (!JwtUtil.validateToken(token)) {
                return Result.error("Invalid token");
            }
            Long userId = JwtUtil.getUserIdFromToken(token);
            List<Recommendation> recommendations = recommendationService.findByUserId(userId);
            return Result.success(recommendations);
        }
        return Result.error("Invalid token");
    } catch (Exception e) {
        logger.error("Failed to get user recommendation list: {}", e.getMessage());
        return Result.error(e.getMessage());
    }
}