diff --git a/CODE_REVIEW.md b/CODE_REVIEW.md new file mode 100644 index 0000000..20ac659 --- /dev/null +++ b/CODE_REVIEW.md @@ -0,0 +1,228 @@ +# 代码审查 + +> 审查范围:`src/main/scala`、`src/main/resources`、`pom.xml`、`data/`(目录约定) +> 审查日期:2026-09-21 + +## 总体评价 + +项目用 Scala 2.11 + Spark 2.4.3 + MySQL(HikariCP + DBUtils)实现了 MovieLens 风格数据集上的三条聚合分析需求,整体结构清晰、能跑通,但存在若干 **正确性 / 安全 / 工程规范** 级别的问题,需要修。 + +下面按严重程度分级。 + +--- + +## 🔴 严重(必须修) + +### 1. MySQL 密码硬编码在源码里 +`src/main/scala/com/example/utils/JDBCUtil.scala:14` + +```scala +private val password = "root123456" +private val url = "jdbc:mysql://192.168.48.10:3306/test" +``` + +- **风险**:密码、IP、库名都明文进 Git;任何克隆者都能拿到内网 MySQL 凭证。 +- **建议**:抽到 `application.conf`(Typesafe Config)或环境变量;至少把密码挪出源码。`dev` 和 `prod` 用不同 profile。 + +### 2. `create_table.sql` 字段注释 copy-paste 错误 +`src/main/resources/create_table.sql:7, 20` + +```sql +`avg_rating` decimal(10, 2) not null comment '电影名称', +... +`avg_rating` DECIMAL(10, 2) NOT NULL COMMENT '电影类别平均评分', +``` + +第 7 行 `top_ten_avg_ratings.avg_rating` 的注释是“电影名称”,明显复制粘贴残留。虽然不影响运行,但生产里这种注释会误导接手人。 + +### 3. `System.exit(-1)` 出现在库代码里 +`JDBCUtil.scala:51` + +```scala +case None => + println("Mysql 连接失败") + System.exit(-1) +``` + +- **问题**:JDBCUtil 应该是个工具对象,不应该决定进程生死。`App.main` 想在 MySQL 不可用时降级(比如只打印到控制台)就没办法。 +- **建议**:抛 `RuntimeException`,让 `main` 顶层 `try/catch` 或让 Spark Driver 自行退出。 + +--- + +## 🟠 主要(建议修) + +### 4. `Entitys.scala` 里的空类 `class Entitys` +`src/main/scala/com/example/spark/Entitys.scala:3` + +```scala +class Entitys { + +} +``` + +完全没人用,是死代码。文件命名也误导(复数 Entitys,里面只有一个空类 + 三个 case class + 一个无关 case class)。建议: + +- 要么删空类,把文件名改成 `Models.scala` 或 `Schemas.scala`; +- 要么拆成 `Movie.scala` / `Rating.scala` / `Results.scala` 三个文件。 + +### 5. 结果 case class 字段类型与 Spark SQL 不一致 +`Entitys.scala:11, 13` + +```scala +case class TopTenAvgRatings(movieId: String, title: String, avgRating: String) +case class TopGenresByAverageRating(genres: String, avgRating: String) +``` + +`Movie.movieId: Int`、`TopGenresByAverageRating` 里 `avgRating` 在 SQL 是 `decimal` / `double`,但这里全是 `String`。原因应该是为了避免 `Int.box/Long.box`,但代价是: + +- 类型擦掉了,编译期查不出 bug; +- `.as[TopTenAvgRatings]` 实际是反射转 String,靠 Spark 的内部转换,调试时容易踩坑; +- 写库时 DBUtils 走 `setObject(String, ...)`,MySQL 驱动能转,但语义不清晰。 + +**建议**:直接用 `Int/Double`,让 Driver 自己装箱;性能差异在这点数据量上完全可忽略。 + +### 6. `insert2Mysql(res: Any)` 用 `Any` 做运行时分发 +`JDBCUtil.scala:44, 56` + +```scala +def insert2Mysql(res: Any): Unit = ... +private def upsert(res: Any, queryRunner: QueryRunner): Unit = { + res match { + case data: TopTenAvgRatings => ... + case data: TopGenresByAverageRating => ... + case data: MovieRatingCount => ... + case _ => println(s"不支持的数据类型:${res.getClass.getName}") + } +} +``` + +- 没有编译期保证,新增 case class 时静默落入 `case _`。 +- `case _` 只 `println` 一下就吞掉,数据丢了不抛错。 + +**建议**:用 trait + 各自实现,例如: + +```scala +sealed trait Writable { def sql: String; def params: Seq[Any] } +object TopTenAvgRatings { def toRow(r: TopTenAvgRatings): Writable = ... } +``` + +或者最简单:拆成 `insertTopTen(Seq[TopTenAvgRatings])`、`insertGenres(...)`、`insertMovieCount(...)`,foreach 在外部调用。 + +### 7. `App.scala` 里的 SQL 注释被去掉执行了 +`App.scala:44, 60-62` + +```scala +// - 需求1:查找电影评分个数超过5000,并且平均分较高的前十部电影名称及其对应的平均评分 +val sql1 = ... +spark.sql(sql1).show() +val result1 = spark.sql(sql1).as[TopTenAvgRatings] +result1.foreach(insert2Mysql(_)) +``` + +- `sql1` 算了两次(一次 `show()`、一次 `as[]`),每次都是从头跑 Job。 +- `show()` 算调试输出,跟"写库"是同一份逻辑,没必要算两遍。 + +**建议**:拿一个 `result1` 用 `.show()` + `.foreach(...)`,或者干脆用 `cache()` 复用。 + +### 8. SQL 3 里 `max(m.movieId)` / `max(m.title)` 是误导写法 +`App.scala:82-91` + +```sql +select + max(m.movieId) movieId +, max(m.title) title +, count(1) cn +from movies m + left join ratings r + on m.movieId = r.movieId + group by m.movieId +``` + +`group by m.movieId` 时 `m.movieId` 已经是 GROUP BY key,`max(m.movieId)` 等于 `m.movieId` 本身;`m.title` 函数依赖 `movieId`,`max(title)` 是合法的但语义上是"随便挑一个"。 + +**建议**: + +```sql +select m.movieId, m.title, count(r.userId) as cn +from movies m left join ratings r on m.movieId = r.movieId +group by m.movieId, m.title +order by cn desc +limit 10 +``` + +更直白。 + +### 9. `SparkSession` 创建后没有 `stop()` +`App.scala:11` + +程序退出靠 JVM 终止,自然回收。但 Driver 日志会一直说 `SparkSession` 没正常关;在 cluster / spark-submit 模式下漏掉 `stop()` 可能影响下一次提交。 + +**建议**: + +```scala +try { ... } finally { spark.stop() } +``` + +### 10. 没有用到的 import +`App.scala:4` + +```scala +import org.apache.spark.sql.{DataFrame, Dataset, SparkSession} +``` + +`DataFrame` 和 `Dataset` 都没引用,IDE 静态检查能识别。可删。 + +### 11. CSV 用 `inferSchema=true` +`App.scala:21-23, 30-32` + +```scala +.option("inferSchema","true") +``` + +每次跑都扫描一次 CSV 推 schema。`movies.csv` 小无所谓,`ratings_all.csv` 如果上 GB 级别,启动慢且 schema 可能飘。 + +**建议**:用 `StructType` 显式声明 schema。 + +--- + +## 🟡 次要(可改可不改) + +| 位置 | 问题 | +|---|---| +| `JDBCUtil.scala:103-105` | `Int.box` / `Long.box` 不必要,DBUtils `update(sql, args*)` 会自动装箱 | +| `JDBCUtil.scala:29` | `HikariDataSource` 是 `object` 字段,永不 close;CLI 应用无所谓,做 library 时会泄漏 | +| `JDBCUtil.scala:33-40` | `getQueryRunner: Option` 模式过度工程化;`dataSource` 在类初始化时已建好,根本不会抛 | +| `App.scala:8-9` | 数据路径硬编码;应该走 args / config | +| `App.scala:14` | `master("local[2]")` 写死;集群提交时得改代码 | +| `log4j.properties` | 只配了 `error` 级别 + stdout,没文件 appender;线上调试不方便 | +| `pom.xml:71-75` | 引了 JUnit 3.8.1,但 `src/test` 整个目录都没有;要么补测试要么删依赖 | +| 缩进风格 | 需求 2/3 里 `val result2/3` 没对齐 `result1`,缩进风格不统一 | +| `sql2` | 没过滤 `category is null or category = '(no genres listed)'`,可能脏数据 | + +--- + +## 改进建议(按优先级) + +1. 把密码 / DB 连接信息挪到 config 文件,加 `.gitignore`。 +2. 删 `class Entitys` 死代码;改文件名。 +3. 修 `create_table.sql` 注释。 +4. 用 sealed trait / 独立函数替换 `Any` 模式匹配;`case _` 改成抛异常。 +5. `sql1` 去重计算 + `spark.stop()` + 显式 CSV schema。 +6. 简化 `TopTenAvgRatings` 等 case class 的字段类型。 +7. 补 1-2 个最关键的单元测试(CSV 解析 / SQL 结果行数)。 + +--- + +## 测试覆盖 + +仓库目前**没有任何测试代码**。`src/test/` 目录是空的,但 `pom.xml` 引了 JUnit 3.8.1。至少建议补: + +- `Entitys` 的 case class 序列化(防 schema 漂移) +- `JDBCUtil.upsert` 至少对一个类型做 round-trip +- 三个 SQL 在 fixture 数据上的预期行数 / Top-1 断言 + +--- + +## 总结 + +作为学习项目骨架是合格的,但**距离"可上线"差上述红/橙清单**。最大的两件事:把密码挪出源码、把空类 / 无用 import / `Any` 分发清理掉。 \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..c4b4094 --- /dev/null +++ b/README.md @@ -0,0 +1,165 @@ +# spark_project + +MovieLens 风格数据集上的三条聚合分析练习项目:基于 Spark SQL + MySQL,把三条统计结果写入数据库。 + +> 学习 / 练手项目;不是生产系统。详见 `CODE_REVIEW.md`。 + +--- + +## 项目简介 + +读取两份 CSV: + +- `data/movies.csv`:电影元数据(`movieId, title, genres`) +- `data/ratings_all.csv`(或 `data/ratings.csv`,小样本):用户评分(`userId, movieId, rating, timestamp`) + +完成三条业务需求: + +| 编号 | 描述 | +| :--: | :-- | +| 需求 1 | 评分次数 > 5000 的电影中,按平均分 Top 10,输出 `movieId / title / avgRating` | +| 需求 2 | 按 `genres` 拆分后,每个类别计算平均评分 | +| 需求 3 | 按评分次数 Top 10 的电影,输出 `movieId / title / ratingCount` | + +三条结果分别写入 MySQL 表 `top_ten_avg_ratings` / `genres_average_rating` / `movie_rating_count`,使用 `REPLACE INTO`(基于各自 unique key 做 upsert)。 + +--- + +## 技术栈 + +| 组件 | 版本 | +| :-- | :-- | +| Scala | 2.11.12 | +| Apache Spark | 2.4.3(Spark Core + Spark SQL)| +| MySQL Connector/J | 8.0.28 | +| HikariCP | 4.0.3 | +| Commons DBUtils | 1.6 | +| 构建工具 | Maven 3.x + `scala-maven-plugin` 4.8.1 | +| Java 目标 | 8 | + +--- + +## 目录结构 + +``` +spark_project/ +├── pom.xml +├── README.md +├── CODE_REVIEW.md +├── data/ +│ ├── movies.csv # 电影元数据 +│ ├── ratings.csv # 小样本评分 +│ └── ratings_all.csv # 全量评分 +└── src/ + ├── main/ + │ ├── resources/ + │ │ ├── create_table.sql # MySQL DDL + │ │ └── log4j.properties # 日志配置(stdout only,error 级别) + │ └── scala/com/example/ + │ ├── spark/ + │ │ ├── App.scala # 入口,三个 SQL 需求 + │ │ └── Entitys.scala # case class:Movie / Rating / 三种结果 + │ └── utils/ + │ └── JDBCUtil.scala # HikariCP + DBUtils 写入 MySQL + └── test/ # 空 +``` + +--- + +## 数据约定 + +- `data/movies.csv` 字段:`movieId, title, genres` + - `genres` 多个类别用 `|` 分隔(如 `Adventure|Animation|Children|Comedy|Fantasy`) +- `data/ratings.csv` / `data/ratings_all.csv` 字段:`userId, movieId, rating, timestamp` +- 默认入口读 `ratings_all.csv`;开发期可在 `App.scala:9` 切回 `ratings.csv` + +--- + +## MySQL 表结构 + +详见 `src/main/resources/create_table.sql`。三张表统一字段:`id` 自增主键、`update_time` 默认当前时间、业务字段 + 唯一键。 + +```sql +-- 需求 1 结果 +create table top_ten_avg_ratings ( + id int(11) not null AUTO_INCREMENT, + movie_id int(11) not null, + title varchar(200) not null, + avg_rating decimal(10, 2) not null, + update_time datetime default current_timestamp, + primary key (id), + unique key movie_id_unique (movie_id) +) ENGINE = InnoDB DEFAULT CHARSET = utf8; +``` + +> ⚠️ 实际库 / 表创建请先用 `create_table.sql` 跑一遍。 + +--- + +## 配置 + +`JDBCUtil.scala` 当前把连接信息硬编码: + +```scala +private val user = "root" +private val password = "root123456" +private val url = "jdbc:mysql://192.168.48.10:3306/test" +``` + +**生产前必须改成外部配置**(application.conf / 环境变量 / 命令行参数)。详见 `CODE_REVIEW.md` 第 1 条。 + +`App.scala:14` 的 `master("local[2]")` 也是硬编码,集群提交前请改为从参数传入。 + +--- + +## 构建与运行 + +### 前置条件 + +- JDK 8 +- Maven 3.6+ +- MySQL 8.x(本地或局域网),能访问 `192.168.48.10:3306` + +### 步骤 + +```powershell +# 1. 建表 +mysql -h 192.168.48.10 -P 3306 -u root -p < src/main/resources/create_table.sql + +# 2. 编译 + 打 fat-jar(assembly 插件已配置) +mvn -P dev clean package + +# 3. dev profile:spark-core / spark-sql 是 compile 作用域,IDE 直接跑 App.main 即可 +# build_pkg profile:provided 作用域,打成可提交到集群的 jar + +# 4. 本地跑(也可直接 java -jar) +java -cp target/spark_project-1.0-SNAPSHOT-jar-with-dependencies.jar com.example.spark.App +``` + +### Maven Profile 说明 + +| profile | 用途 | Spark 依赖 scope | +| :-- | :-- | :-- | +| `dev` | IDE 本地跑、单元测试 | `compile` | +| `build_pkg` | 打胖包丢到集群跑 | `provided` | + +--- + +## 已知问题 / TODO + +详见 [`CODE_REVIEW.md`](CODE_REVIEW.md) 完整清单,优先级最高的: + +- 🔴 MySQL 密码硬编码 +- 🔴 `create_table.sql` 中 `avg_rating` 字段注释残留 +- 🟠 `Entitys.scala` 死类 + 字段类型与 SQL 不一致 +- 🟠 `insert2Mysql(res: Any)` 用 Any 做运行时分发 +- 🟠 `SparkSession` 未 `stop()`;CSV `inferSchema=true` +- 🟡 无单元测试(JUnit 依赖已引但 `src/test/` 空) + +--- + +## 参考 + +- MovieLens 数据集: +- Spark SQL 文档(2.4.x): +- HikariCP 配置: \ No newline at end of file