Spark 学习项目

仓库：KennethCheng/spark_20260915

我的 Scala + Java 双实现 Spark 学习项目，统一放在一个 Maven 工程里。代码分两条线：

• chapter —— 线下课程 Demo（org.example.*）：核心算子 / 分区 / 缓存 / Checkpoint / Spark SQL / Spark On Hive
• practise —— ChatGPT 课程练习（org.exercises.exercisesN.*）：每课一目录，含 demo/ 演示 + exercises/ 练习题

主线按 Spark 2.4.3 / Scala 2.11 / JDK 8 学习，从 RDD 入门到 Spark SQL / Hive / JDBC / Yarn 提交。

---

⚠️ 安全提醒（先看）

• src/main/resources/hive-site.xml 与 src/main/scala/org/example/sql/*.scala 中含有明文 MySQL 账号密码与内网 JDBC URL（参见审查报告）
• 当前仓库是公开的 Gitea，凭据已暴露在 git 历史中，先轮换密码再继续
• AGENTS.md 已要求凭据外部化（环境变量 / *.local.properties + .gitignore），本次重写 README 不修代码，仅做记录

---

技术栈

类别	版本
JDK	8
Scala	2.11.12
Apache Spark	2.4.3（core / sql / hive）
Hadoop Client	2.7.3
Maven	Maven 3
scala-maven-plugin	4.8.1
maven-assembly-plugin	3.7.1
MySQL Connector	8.0.28
HikariCP	4.0.3
commons-dbutils	1.6
JUnit	3.8.1（test scope，当前无用例）

---

项目结构

spark_project/
├── pom.xml
├── AGENTS.md                        # Agent / 协作规约（必读）
├── README.md
│
├── src/main/java/org/example/       # Java 版核心 Demo
│   ├── AppTest.java
│   └── core/
│       ├── GroupByKeyDemo.java
│       ├── MyPartitioner.java
│       ├── RddCreateDemo.java
│       └── RepartitionVsCoaleasce.java
│
├── src/main/scala/org/example/      # Scala 版核心 Demo（与 Java 同名 4 个 + 扩展）
│   ├── App.scala                    # local[2] 入口：word.txt WordCount
│   ├── App2.scala                   # spark-submit 集群提交示例（注释里有完整命令）
│   ├── test.scala
│   ├── core/
│   │   ├── GroupByKeyDemo.scala
│   │   ├── MyPartitioner.scala
│   │   ├── RddCreateDemo.scala
│   │   ├── RepartitionVsCoalesce.scala
│   │   ├── CacheTest.scala
│   │   ├── DemoCache.scala          # cache + persist 性能对比
│   │   ├── DemoCacheNoCache.scala
│   │   ├── CheckpointDemo.scala     # setCheckpointDir + checkpoint()
└── ReadFromCheckpoint.scala
│   └── sql/                         # Spark SQL 线下课 Demo
│       ├── CreateDataFrame.scala    # RDD+Schema / CSV / JDBC 三种方式
│       ├── DataFrameDSL.scala       # select / col / as
│       ├── DataFrameSql.scala       # createTempView + spark.sql
│       └── SparkOnHive.scala        # enableHiveSupport + spark.sql
│
├── src/main/scala/org/exercises/    # ChatGPT 课程练习（按课号递增）
│   ├── exercises1/                  # 第 1 课：Spark 入门
│   │   ├── demo/   (DemoS1tart / Demo2CreateRDD / Demo3RddMap)
│   │   └── exercises/ (Exercises1~5)
│   ├── exercises2/                  # 第 2 课：RDD 算子 map / filter / flatMap / reduce / foreach
│   ├── exercises3/                  # 第 3 课：PairRDD / reduceByKey / WordCount
│   ├── exercises4/                  # 第 4 课：groupByKey vs reduceByKey / Shuffle
│   ├── exercises5/                  # 第 5 课：Partition / partitionBy / 自定义 Partitioner
│   ├── exercises6/                  # 第 6 课：窄依赖 / 宽依赖 / Stage / Task
│   ├── exercises7/                  # 第 7 课：cache / persist / mapPartitions
│   └── exercises8/                  # 第 8 课：cache / persist / checkpoint
│
└── src/main/resources/
    ├── create_table.sql             # 离线分析结果目标表（top_ten_avg_ratings / genres_average_rating）
    ├── hive-site.xml                # ⚠️ 含明文凭据，待外部化
    ├── log4j.properties
    ├── word.txt                     # WordCount 示例输入
    └── spark/sql/student.csv        # DataFrame 读取示例

src/main/resources/tmp/spark-checkpoint/ 是运行时产物，已被运行时写入但不应进仓库，后续会加入 .gitignore。

---

ChatGPT 课程进度

课号	主题	目录	演示 Demo	状态
1	Spark 入门（Conf / Context / parallelize / textFile / Transformation vs Action）	exercises1	DemoS1tart、Demo2CreateRDD、Demo3RddMap	✅
2	RDD 算子：map / filter / flatMap / reduce / foreach	exercises2	Demo1Map ~ Demo5Foreach	✅
3	PairRDD / mapValues / reduceByKey / groupByKey / WordCount	exercises3	Demo1Map ~ Demo5WordCount	✅
4	groupByKey vs reduceByKey / Shuffle	exercises4	Demo1GroupByKeyAndReduceByKey、Demo2GroupByKeyVsReduceByKey	✅
5	Partition / partitionBy / 自定义 Partitioner / repartition / coalesce	exercises5	Demo1PartitionBasic ~ Demo4MyPartitioner	✅
6	窄依赖 / 宽依赖 / Stage / Task / DAG	exercises6	Demo.scala	✅
7	Cache / Persist / mapPartitions / 性能对比	exercises7	Demo1CacheAndPersist、Demo2MapPartitions	✅
8	cache / persist / setCheckpointDir / checkpoint 复用	exercises8	Demo1Checkpoint	✅

每课 5 道练习，统一以 Exercises1.scala ~ Exercises5.scala 命名，可独立 main 运行。

---

线下课 Demo 一览（org.example.*）

包	Demo	说明
org.example	App	local[2] 跑 word.txt 的 WordCount 入门入口
org.example	App2	集群 spark-submit 模板，注释里写好了完整提交命令
org.example.core	RddCreateDemo	parallelize + textFile 两种 RDD 创建方式
org.example.core	GroupByKeyDemo	groupByKey 行为与输出观察
org.example.core	RepartitionVsCoalesce	repartition vs coalesce 区别
org.example.core	MyPartitioner	自定义 Partitioner
org.example.core	DemoCache / DemoCacheNoCache / CacheTest	cache / persist(MEMORY_ONLY) 是否真的更快
org.example.core	CheckpointDemo / ReadFromCheckpoint	setCheckpointDir + 复用 checkpoint 切断 lineage
org.example.sql	CreateDataFrame	RDD+Schema / CSV / JDBC 三种建 DF 方式
org.example.sql	DataFrameDSL	DSL：select / col / as
org.example.sql	DataFrameSql	createTempView + spark.sql
org.example.sql	SparkOnHive	enableHiveSupport() + spark.sql("use gmall")
org.example	test.scala	随手调试用，可忽略

org.example.core.* 在 src/main/java 下还有 4 个 Java 版同名 Demo，方便对照两种语言写法。

---

当前已掌握的 RDD 知识体系

Spark
├── SparkConf / SparkContext / SparkSession
│
├── RDD
│   ├── 创建
│   │   ├── parallelize(makeRDD)
│   │   └── textFile
│   │
│   ├── Transformation（Lazy）
│   │   ├── map / mapPartitions
│   │   ├── filter
│   │   ├── flatMap
│   │   ├── groupBy / sortBy / sortWith
│   │   └── reduceByKey / groupByKey
│   │
│   ├── Action（触发 Job）
│   │   ├── collect / count / take
│   │   ├── reduce / foreach
│   │   └── saveAsTextFile
│   │
│   ├── Partition
│   │   ├── getNumPartitions / glom
│   │   ├── partitionBy(自定义 Partitioner)
│   │   ├── repartition  ← 触发 Shuffle
│   │   └── coalesce      ← 默认不 Shuffle
│   │
│   └── 性能
│       ├── Lazy Evaluation
│       ├── cache / persist(StorageLevel)
│       └── checkpoint（切断 lineage，依赖可靠文件系统）
│
└── PairRDD
    ├── reduceByKey / foldByKey
    ├── groupByKey
    └── 自定义 Partitioner

---

Spark SQL / DataFrame / Hive

RDD 基础阶段完成后，开始 Spark SQL：

• SparkSession.builder().enableHiveSupport().getOrCreate()
• DataFrame 创建：RDD+Schema / spark.read.csv(...) / spark.read.format("jdbc")
• DSL：select / col / as
• SQL：createTempView + spark.sql(...)
• Hive：spark.sql("use gmall") + 聚合查询
• 目标：DataFrame 聚合结果 → JDBC → MySQL（建表 SQL 在 create_table.sql）

---

Spark 集群实践

项目	说明
spark-submit	App2.scala 注释里给出完整 yarn-client 提交命令
Spark UI	本地 local[2] 跑起来后 http://localhost:4040（Jobs / Stages / Storage / SQL）
History Server	18080（仅历史任务）
Yarn Client	--master yarn --deploy-mode client
HDFS	textFile("hdfs://...") 见 RddCreateDemo 注释
Hive / Beeline / DBeaver	已跑通 enableHiveSupport 链路

注意：

• 4040 是当前 Application 的实时 UI，结束即失效
• 18080 是 History Server，两者不是一回事

---

快速开始

1. 环境

JDK 8、Scala 2.11.x、Maven 3.x，本地能 mvn 即可。

2. 拉依赖与编译

mvn -DskipTests dependency:resolve
mvn compile

3. 打包（可选）

mvn package

产物：

target/spark_project-1.0-SNAPSHOT-jar-with-dependencies.jar

4. 跑单个 Demo

任选其一：

• IDE 里把目标 object ExercisesX / object Xxx / class Xxx 当作 main 直接 Run
• 命令行：

java -cp target/spark_project-1.0-SNAPSHOT-jar-with-dependencies.jar org.example.App

集群提交参考 App2.scala 顶部注释。

---

MySQL / Hive 配置

• JDBC 表结构：src/main/resources/create_table.sql
  • top_ten_avg_ratings —— Top10 平均评分电影
  • genres_average_rating —— 类型平均评分
• Hive：src/main/resources/hive-site.xml（⚠️ 内含明文凭据，待外部化）
• 依赖：mysql-connector-java / HikariCP / commons-dbutils

凭据外部化建议（参考 AGENTS.md）：

src/main/resources/
├── application.properties.example   # 模板：仅占位符，提交到仓库
└── application.local.properties     # 真凭据，gitignore

读取：
  val props = new Properties()
  props.load(getClass.getResourceAsStream("/application.properties"))
  props.load(getClass.getResourceAsStream("/application.local.properties"))

---

代码规范与提交流程

• 单文件尽量一个知识点，命名沿用 DemoXxx / ExercisesX / XxxDemo
• 中文注释 + 中文 commit message
• 推荐样式：

feat:线下课：集合_xxx
feat:ChatGPT 第 7 课：缓存 Cache / Persist + mapPartitions
fix:修复 ...
docs:更新 README
refactor:重构 ...

• 单课 / 单练习保持原子提交
• main 分支受保护，从 main 拉 feature 分支再 PR（详见 AGENTS.md）

---

学习路线

Scala 基础
  ↓
RDD 入门（parallelize / textFile / Transformation vs Action）
  ↓
RDD 算子（map / filter / flatMap / reduce / foreach）
  ↓
PairRDD（reduceByKey / groupByKey / WordCount）
  ↓
Partition / Shuffle / 自定义 Partitioner
  ↓
依赖 / DAG / Stage / Task
  ↓
Cache / Persist / Checkpoint
  ↓
Spark SQL（DataFrame / DSL / SQL）
  ↓
Hive（enableHiveSupport / Metastore）
  ↓
JDBC 读写 MySQL
  ↓
Spark Thrift Server / Beeline / DBeaver
  ↓
Yarn / spark-submit / 集群运行
  ↓
完整数据分析项目

---

相关资源

• Apache Spark：https://spark.apache.org/
• Apache Hadoop：https://hadoop.apache.org/
• Apache Maven：https://maven.apache.org/
• Hive 配置文档：https://cwiki.apache.org/confluence/display/Hive/Configuration+Properties