diff --git a/README.md b/README.md index e2f5c1c..ad1a93f 100644 --- a/README.md +++ b/README.md @@ -2,27 +2,20 @@ > 仓库:[`KennethCheng/spark_20260915`](https://gitea.kennethcheng.cc/KennethCheng/spark_20260915) -这是我的 **Scala + Spark 学习与练习项目**,主要用于记录跟随课程学习 Spark 的过程,并把课堂 Demo、课后练习和实际运行实验整理到同一个 Maven 项目中。 +我的 **Scala + Java 双实现 Spark 学习项目**,统一放在一个 Maven 工程里。代码分两条线: -当前学习重点以 **Spark Core / RDD** 为主,并逐步过渡到 **Spark SQL / DataFrame / Hive / JDBC**。 +- `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 提交。 --- -## 学习方式 +## ⚠️ 安全提醒(先看) -项目中的代码主要分成两类: - -- `chapter`:线下课程内容、课堂 Demo、知识点验证 -- `practise`:配合 ChatGPT 学习完成的课后练习 - -代码统一放在 Maven 的 `src/main/scala` / `src/main/java` 下,尽量保持一个知识点一个 Demo,方便单独运行和复习。 - -提交信息统一使用中文,例如: - -```text -feat:线下课:集合_... -feat:ChatGPT第 7 课:缓存 Cache / Persist + mapPartitions -``` +- `src/main/resources/hive-site.xml` 与 `src/main/scala/org/example/sql/*.scala` 中含有**明文 MySQL 账号密码**与内网 JDBC URL(参见审查报告) +- 当前仓库是公开的 Gitea,凭据已暴露在 git 历史中,**先轮换密码再继续** +- AGENTS.md 已要求凭据外部化(环境变量 / `*.local.properties` + `.gitignore`),本次重写 README 不修代码,仅做记录 --- @@ -30,9 +23,9 @@ feat:ChatGPT第 7 课:缓存 Cache / Persist + mapPartitions | 类别 | 版本 | | --- | --- | -| JDK | Java 8 | +| JDK | 8 | | Scala | 2.11.12 | -| Apache Spark | 2.4.3 | +| Apache Spark | 2.4.3(core / sql / hive) | | Hadoop Client | 2.7.3 | | Maven | Maven 3 | | scala-maven-plugin | 4.8.1 | @@ -40,8 +33,7 @@ feat:ChatGPT第 7 课:缓存 Cache / Persist + mapPartitions | MySQL Connector | 8.0.28 | | HikariCP | 4.0.3 | | commons-dbutils | 1.6 | - -> 项目以 **Spark 2.4.3 + Scala 2.11 + JDK 8** 为基准,运行环境不要随意升级 JVM / Scala 版本。 +| JUnit | 3.8.1(test scope,当前无用例) | --- @@ -50,250 +42,97 @@ feat:ChatGPT第 7 课:缓存 Cache / Persist + mapPartitions ```text spark_project/ ├── pom.xml -├── AGENTS.md +├── AGENTS.md # Agent / 协作规约(必读) ├── README.md │ -├── src/main/java/org/example/ -│ ├── core/ -│ │ ├── GroupByKeyDemo.java -│ │ ├── MyPartitioner.java -│ │ ├── RddCreateDemo.java -│ │ └── RepartitionVsCoalesce.java -│ └── AppTest.java -│ -├── src/main/scala/org/example/ -│ ├── App.scala -│ ├── test.scala +├── src/main/java/org/example/ # Java 版核心 Demo +│ ├── AppTest.java │ └── core/ -│ ├── GroupByKeyDemo.scala -│ ├── MyPartitioner.scala -│ ├── RddCreateDemo.scala -│ └── RepartitionVsCoalesce.scala +│ ├── GroupByKeyDemo.java +│ ├── MyPartitioner.java +│ ├── RddCreateDemo.java +│ └── RepartitionVsCoaleasce.java │ -├── src/main/scala/org/exercises/ -│ ├── exercises1/ -│ ├── exercises2/ -│ ├── exercises3/ -│ └── ... +├── 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 + ├── create_table.sql # 离线分析结果目标表(top_ten_avg_ratings / genres_average_rating) + ├── hive-site.xml # ⚠️ 含明文凭据,待外部化 ├── log4j.properties - └── word.txt + ├── word.txt # WordCount 示例输入 + └── spark/sql/student.csv # DataFrame 读取示例 ``` +> `src/main/resources/tmp/spark-checkpoint/` 是运行时产物,已被运行时写入但**不应进仓库**,后续会加入 `.gitignore`。 + --- -## Spark 学习进度 +## ChatGPT 课程进度 -### 第 1 课:Spark 入门 +| 课号 | 主题 | 目录 | 演示 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` | ✅ | -主要内容: - -- `SparkConf` -- `SparkContext` -- `parallelize` -- 从文件创建 RDD -- Spark 程序基本运行方式 -- Transformation / Action 基本概念 - -对应练习: - -```text -src/main/scala/org/exercises/exercises1/ -``` +每课 5 道练习,统一以 `Exercises1.scala` ~ `Exercises5.scala` 命名,可独立 `main` 运行。 --- -### 第 2 课:RDD 常用算子 +## 线下课 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` | 随手调试用,可忽略 | -- `map` -- `filter` -- `flatMap` -- `reduce` -- `foreach` - -重点理解: - -```text -Transformation - ↓ -生成新的 RDD - ↓ -继续构建计算链 - -Action - ↓ -触发 Job 执行 - ↓ -返回结果 / 输出结果 -``` - -对应练习: - -```text -src/main/scala/org/exercises/exercises2/ -``` - ---- - -### 第 3 课:PairRDD / WordCount - -主要学习: - -- `(K, V)` 键值对 RDD -- `map` 生成 PairRDD -- `reduceByKey` -- WordCount -- `groupByKey` -- shuffle 的基本概念 - -典型流程: - -```text -文本 - ↓ -flatMap - ↓ -map - ↓ -reduceByKey - ↓ -collect -``` - -对应练习: - -```text -src/main/scala/org/exercises/exercises3/ -``` - ---- - -### 第 4~5 课:Scala 函数基础 + Spark 分区 - -Scala 部分重点: - -- `def` -- `=>` -- `map` -- 函数作为参数 -- `_` 下划线简写 -- Lambda / 函数值 -- 函数参数与返回值 - -Spark 部分重点: - -- Partition -- `getNumPartitions` -- `glom` -- `mapPartitions` -- `partitionBy` -- 自定义 `Partitioner` -- `groupByKey` -- `reduceByKey` -- `repartition` -- `coalesce` - -项目中的核心 Demo: - -```text -src/main/scala/org/example/core/ -src/main/java/org/example/core/ -``` - ---- - -### 第 6~7 课:RDD 执行过程与缓存 - -主要内容: - -- Transformation 与 Action -- Lazy Evaluation -- DAG -- Job / Stage / Task -- Cache -- Persist -- `mapPartitions` -- 多次 Action 时为什么缓存可能提升性能 -- Spark UI 中查看 Job / Stage / Task - -重点认识: - -```text -Transformation - ↓ -只记录计算逻辑 - ↓ -Action - ↓ -真正触发 Job -``` - -以及: - -```text -RDD - ├── cache() - └── persist(...) -``` - -缓存不是“加了就一定更快”,只有同一个 RDD 被重复使用、重新计算成本较高时,收益才比较明显。 - ---- - -### 第 8 课:flatMap 与嵌套集合 - -主要学习: - -- `map` 与 `flatMap` 的区别 -- 嵌套集合处理 -- 一对多转换 -- WordCount 中为什么常用 `flatMap` -- 不使用 `flatMap` 的替代实现 - -典型理解: - -```text -map -1 -> List(1, 2) -2 -> List(3, 4) - -结果: -List(List(1, 2), List(3, 4)) -``` - -```text -flatMap -1 -> List(1, 2) -2 -> List(3, 4) - -结果: -List(1, 2, 3, 4) -``` - ---- - -### 第 9 课:排序与分组 - -主要学习: - -- `sortBy` -- `sortWith` -- `groupBy` -- 分组后 `Iterable / CompactBuffer` 的理解 -- 排序与分组的使用场景 -- PairRDD 与普通 RDD 的进一步操作 - -常见写法: - -```scala -rdd.sortBy(_._2) -rdd.sortWith((a, b) => a._2 > b._2) -rdd.groupBy(_ % 2) -``` +> `org.example.core.*` 在 `src/main/java` 下还有 4 个 Java 版同名 Demo,方便对照两种语言写法。 --- @@ -301,44 +140,38 @@ rdd.groupBy(_ % 2) ```text Spark -│ -├── SparkConf / SparkContext +├── SparkConf / SparkContext / SparkSession │ ├── RDD │ ├── 创建 -│ │ ├── parallelize +│ │ ├── parallelize(makeRDD) │ │ └── textFile │ │ -│ ├── Transformation -│ │ ├── map +│ ├── Transformation(Lazy) +│ │ ├── map / mapPartitions │ │ ├── filter │ │ ├── flatMap -│ │ ├── mapPartitions -│ │ ├── groupBy -│ │ ├── sortBy -│ │ ├── sortWith -│ │ └── reduceByKey +│ │ ├── groupBy / sortBy / sortWith +│ │ └── reduceByKey / groupByKey │ │ -│ ├── Action -│ │ ├── collect -│ │ ├── count -│ │ ├── reduce -│ │ └── foreach +│ ├── Action(触发 Job) +│ │ ├── collect / count / take +│ │ ├── reduce / foreach +│ │ └── saveAsTextFile │ │ │ ├── Partition -│ │ ├── glom -│ │ ├── getNumPartitions -│ │ ├── partitionBy -│ │ ├── repartition -│ │ └── coalesce +│ │ ├── getNumPartitions / glom +│ │ ├── partitionBy(自定义 Partitioner) +│ │ ├── repartition ← 触发 Shuffle +│ │ └── coalesce ← 默认不 Shuffle │ │ │ └── 性能 -│ ├── cache -│ ├── persist -│ └── Lazy Evaluation +│ ├── Lazy Evaluation +│ ├── cache / persist(StorageLevel) +│ └── checkpoint(切断 lineage,依赖可靠文件系统) │ └── PairRDD - ├── reduceByKey + ├── reduceByKey / foldByKey ├── groupByKey └── 自定义 Partitioner ``` @@ -347,261 +180,147 @@ Spark ## Spark SQL / DataFrame / Hive -RDD 基础阶段完成后,项目继续进入 Spark SQL 相关内容: +RDD 基础阶段完成后,开始 Spark SQL: -- `SparkSession` -- DataFrame -- DataSet 基础概念 -- Schema -- DSL 查询 -- SQL 查询 -- JDBC -- Hive 支持 -- Spark Thrift Server -- DBeaver 连接 Spark SQL -- DataFrame 与 JDBC 数据写入 - -目前已经开始接触: - -```scala -val spark = SparkSession.builder() - .appName("Spark On Hive") - .enableHiveSupport() - .getOrCreate() -``` - -以及: - -```scala -spark.sql("show databases") -spark.sql("show tables") -spark.sql("select ...") -``` - -后续会把 RDD 学到的思想逐步对应到 DataFrame / 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 集群实践 -除了本地 `local` 模式,还会进行 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` 链路 | -已经实践过的内容包括: +注意: -- `spark-submit` -- Yarn Client -- HDFS -- Spark UI -- Spark History Server -- Hive / Hive Metastore -- Spark Thrift Server -- Beeline -- DBeaver -- Spark SQL - -例如: - -```bash -spark-submit \ - --master yarn \ - --deploy-mode client \ - --class org.example.core.App2 \ - spark_project-1.0-SNAPSHOT.jar -``` +- `4040` 是当前 Application 的实时 UI,结束即失效 +- `18080` 是 History Server,两者不是一回事 --- ## 快速开始 -### 1. 准备环境 +### 1. 环境 -确保本机具备: +JDK 8、Scala 2.11.x、Maven 3.x,本地能 `mvn` 即可。 -```text -JDK 8 -Scala 2.11.x -Maven 3.x -``` - -### 2. 编译 +### 2. 拉依赖与编译 ```bash mvn -DskipTests dependency:resolve - mvn compile ``` -### 3. 打包 +### 3. 打包(可选) ```bash mvn package ``` -生成: +产物: ```text target/spark_project-1.0-SNAPSHOT-jar-with-dependencies.jar ``` -### 4. IDEA 运行 +### 4. 跑单个 Demo -在 IntelliJ IDEA 中直接运行 Scala / Java 中的 `object` 或 `main` 方法即可。 +任选其一: + +- IDE 里把目标 `object ExercisesX` / `object Xxx` / `class Xxx` 当作 `main` 直接 Run +- 命令行: + +```bash +java -cp target/spark_project-1.0-SNAPSHOT-jar-with-dependencies.jar org.example.App +``` + +集群提交参考 `App2.scala` 顶部注释。 --- -## 本地 Spark UI +## MySQL / Hive 配置 -本地运行 Spark 程序时,可以通过 Spark Web UI 查看: +- 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): ```text -http://localhost:4040 -``` +src/main/resources/ +├── application.properties.example # 模板:仅占位符,提交到仓库 +└── application.local.properties # 真凭据,gitignore -常见页面: - -```text -Jobs -Stages -Storage -Environment -Executors -SQL -``` - -注意: - -- `4040` 是当前 Spark Application 的 Web UI 端口 -- Application 结束后,4040 通常就不再提供该 Application 的实时页面 -- `18080` 是 Spark History Server,不是本地运行 Application 的实时 UI - ---- - -## MySQL / JDBC - -`src/main/resources/create_table.sql` 中保留了后续分析结果的目标表。 - -目前主要规划: - -```text -Spark DataFrame - ↓ - 聚合分析 - ↓ - JDBC 写入 - ↓ - MySQL -``` - -目标表: - -| 表名 | 用途 | -| --- | --- | -| `top_ten_avg_ratings` | Top10 平均评分电影 | -| `genres_average_rating` | 类型平均评分 | - -相关依赖: - -```text -mysql-connector-java -HikariCP -commons-dbutils +读取: + val props = new Properties() + props.load(getClass.getResourceAsStream("/application.properties")) + props.load(getClass.getResourceAsStream("/application.local.properties")) ``` --- -## 代码规范 +## 代码规范与提交流程 -### Demo 命名 - -按课程 / 知识点拆分 Demo,尽量保持: +- 单文件尽量一个知识点,命名沿用 `DemoXxx` / `ExercisesX` / `XxxDemo` +- 中文注释 + **中文 commit message** +- 推荐样式: ```text -Demo1xxx -Demo2xxx -... -``` - -### Commit - -推荐: - -```text -feat:线下课:集合_... -feat:ChatGPT第 7 课:缓存 Cache / Persist + mapPartitions +feat:线下课:集合_xxx +feat:ChatGPT 第 7 课:缓存 Cache / Persist + mapPartitions fix:修复 ... docs:更新 README refactor:重构 ... ``` -每个知识点尽量保持一个独立 commit,便于后续按课程回看。 +- 单课 / 单练习保持原子提交 +- `main` 分支受保护,从 `main` 拉 feature 分支再 PR(详见 AGENTS.md) --- -## 学习目标 - -最终形成一套完整的 Spark 学习路线: +## 学习路线 ```text Scala 基础 - ↓ -RDD 基础 - ↓ -PairRDD - ↓ -Partition / Shuffle - ↓ -Cache / DAG / Stage / Task - ↓ -Spark SQL / DataFrame - ↓ -Hive - ↓ -JDBC - ↓ -Spark Thrift Server - ↓ -Yarn / 集群运行 - ↓ + ↓ +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 / 集群运行 + ↓ 完整数据分析项目 ``` -重点不是只记 API,而是逐步理解: - -```text -代码怎么写 - ↓ -RDD 怎么产生 - ↓ -Transformation 怎么串起来 - ↓ -Action 什么时候触发 - ↓ -Spark 如何拆成 Job / Stage / Task - ↓ -数据如何在 Partition 之间流动 - ↓ -Shuffle 为什么发生 - ↓ -为什么某些写法性能更好 -``` - ---- - -## 学习记录 - -| 阶段 | 内容 | 状态 | -| --- | --- | --- | -| 01 | Spark 入门 | ✅ | -| 02 | RDD 常用算子 | ✅ | -| 03 | PairRDD / WordCount | ✅ | -| 04 | Scala 函数基础 | ✅ | -| 05 | 函数作为参数 / Partition | ✅ | -| 06 | RDD 执行与惰性计算 | ✅ | -| 07 | Cache / Persist / mapPartitions | ✅ | -| 08 | flatMap / 嵌套集合 | ✅ | -| 09 | sortBy / sortWith / groupBy | ✅ | -| 10+ | Spark SQL / DataFrame / Hive | 🚧 | - --- ## 相关资源 @@ -609,3 +328,4 @@ Shuffle 为什么发生 - Apache Spark: - Apache Hadoop: - Apache Maven: +- Hive 配置文档: \ No newline at end of file