真问-1 - gcd

1、iceberg 的文件层级结构

iceberg 表是一条从 catalog 一路指到数据文件的指针链，每一层都是为了少一次对象存储的 list；

flowchart TD
    A["Catalog<br/>(表名 → 当前 metadata 指针)"] --> B["metadata.json<br/>(表当前状态: schema/分区/快照列表/current-snapshot-id)"]
    B --> C["Snapshot<br/>(表某一刻的完整状态)"]
    C --> D["Manifest List<br/>(snap-*.avro: 这份快照下所有 manifest 的清单)"]
    D --> E["Manifest<br/>(manifest-*.avro: 数据文件清单 + 每文件的统计)"]
    E --> F["数据文件<br/>(*.parquet / *.avro / *.orc)"]
    E -.also.-> G["删除文件<br/>(position/equality deletes, v2)"]
    B -.历史.-> H["metadata-log<br/>(旧 metadata.json 列表)"]
    C -.历史.-> I["snapshot-log<br/>(旧快照变更记录)"]

层	物理形态	存什么	为什么单独成层
Catalog	一个外部服务/表（Hive Metastore / JDBC 表 / REST 服务 / Nessie）	表名 → 当前 metadata 文件的位置指针	给「表」一个稳定入口。Catalog 不存数据，只存「这张表现在指向哪个 metadata」。换快照就是改这个指针
metadata.json	表根目录下一个 JSON 文件（`v1.metadata.json` / `v2.metadata.json`，带版本号递增）	schema（字段定义）、partition-spec（分区规则）、sort-order、properties、所有快照列表、`current-snapshot-id`、`last-sequence-number`、metadata-log、snapshot-log	表的「当前状态」快照。读表第一步就是拿它，知道现在该看哪个快照
Snapshot	metadata.json 里的一条记录（不是独立文件）	指向一个 manifest list、父快照 id、操作类型（append/overwrite/replace）、时间戳、变更统计（加/删了多少文件、多少行）	一次写操作 = 一个新快照。time travel 的基本单位
Manifest List	一个 Avro 文件（`snap-<id>-<attempt>.avro`）	指向多个 manifest + 每个 manifest 的统计（文件数、记录数、分区范围）	一份快照下可能有很多 manifest，这一层是「manifest 的目录页」，让读端能按分区范围跳过整个 manifest
Manifest	一个 Avro 文件（`<attempt>-manifest-<id>.avro`）	每个数据文件的：路径、格式、分区值、记录数、文件大小、每列的 min/max/null 计数	数据文件的索引页。读端用它做分区裁剪 + 文件裁剪，不读不必要的数据文件
数据文件	Parquet/Avro/ORC	真正的行数据	实际被计算引擎读的文件
删除文件（v2）	position delete（行号）/ equality delete（键值）	标记哪些行被删/被改	v2 规范支持行级 update/delete，不重写整文件，只追加「删除标记」

对象存储（S3/OSS）有两个要命特性：① list 操作又慢又贵（高延迟、有速率限制，且文件越多越慢，是 O(文件数) 次往返）；② 没有目录，只有 key 前缀模拟目录。

如果像 Hive 那样把「哪些文件属于这张表」靠目录结构 + list 来发现，文件一多查询规划就崩。Iceberg 的解法是把「哪些文件存在」记在 manifest 里——manifest 是一个普通对象，一次 GET 拿到全量文件清单（含统计），不用 list 目录。于是：

查询规划变成「读几个 manifest 文件」而不是「list 几千个数据文件」。
快照切换只改 metadata 里的指针（current-snapshot-id），不动任何数据文件——这是原子提交和 time travel 的物理基础。
分区裁剪、文件裁剪靠 manifest 里的分区值和 min/max 统计，在读数据前就完成。

代价：层级多 = 元数据本身要维护。写一次表要更新 manifest → manifest list → snapshot → metadata，且每层都可能产生小文件

2、以文件层级结构的视角，写入一个 iceberg 表的流程

第 0 步：拿到当前表的基线

Writer 向 Catalog 查表 → 拿到当前 metadata.json 的位置。
读 metadata.json → 得到 current-snapshot-id、当前 partition-spec、schema。
这一步决定了「我要在哪个快照之上追加」。

第 1 步：写数据文件

Writer 把这批数据按 partition-spec 分区，写成若干 Parquet 文件，落到表的数据目录下（按分区值摆目录，如 data/event_date=2026-09-14/）。
关键：这一步只写文件，不碰任何元数据，也不删老文件。 Iceberg 的写是纯追加。

第 2 步：写新 manifest

为这批新数据文件生成一个（或几个）manifest 文件，记录每个数据文件的路径、大小、记录数、分区值、列统计。
这个 manifest 是「这批新文件的索引页」。

第 3 步：组装新 manifest list

把「新写的 manifest」+「当前快照已有的老 manifest」合并成一个新的 manifest list 文件。
注意是合并清单，不是合并数据——老 manifest 原封不动地被新 manifest list 引用。

第 4 步：生成新快照 + 新 metadata.json

在 metadata.json 里新增一条 snapshot：指向新 manifest list，父快照是当前快照，操作类型 append，记录加了几个文件多少行。
current-snapshot-id 指向新快照。旧 metadata 进 metadata-log，旧快照记录进 snapshot-log。

第 5 步：原子提交（commit）

这是唯一有并发风险的步骤。Writer 拿着「我基于的 metadata 版本」去 Catalog 做乐观并发提交：Catalog 检查「当前 metadata 指针」是不是还是我基于的那个版本——
- 是 → 把指针 CAS 换成新 metadata.json，提交成功。
- 不是 → 说明有人先提交了，我冲突了。Writer 重新读最新 metadata，把自己的改动 merge 上去（重做第 3、4 步），再试。

sequenceDiagram
    participant W as Writer
    participant C as Catalog
    participant S3 as 对象存储
    W->>C: 查表 → 当前 metadata 指针
    C-->>W: metadata_v5.json
    W->>S3: 1.写数据文件 *.parquet
    W->>S3: 2.写新 manifest
    W->>S3: 3.写新 manifest list (合并老 manifest)
    W->>S3: 4.写 metadata_v6.json (current→新快照)
    W->>C: 5.提交: CAS(metadata_v5 → metadata_v6)
    alt 无冲突
        C-->>W: 成功
    else 有冲突
        C-->>W: 冲突,当前已是 v6
        W->>C: 重读 v6, merge, 重试
    end

为什么 commit 只换指针：因为数据文件、manifest、manifest list 都是不可变的（immutable），写完就不改。唯一可变的是「metadata 指针指向谁」。把一个复杂的多文件原子操作压缩成「换一个指针」，这是 Iceberg 在对象存储上实现 ACID 的关键——对象存储本身不支持跨文件事务，但支持单对象的写。

3、以文件层级结构的视角，读取一个 iceberg 表的流程

读表 = 沿指针链一路下钻，在每一层做裁剪，只读真正需要的数据文件。这是 Iceberg 查询快的根因。

第 1 步：定位当前快照

Reader 向 Catalog 查表 → 拿到当前 metadata.json。
读 metadata.json → current-snapshot-id → 找到当前快照 → 拿到 manifest list 指针。

第 2 步：读 manifest list，做 manifest 级裁剪

读 manifest list 文件，拿到所有 manifest 及其分区范围统计。
用查询的分区过滤条件（如 event_date = '2026-09-14'）跳过分区范围不匹配的 manifest——整个 manifest 直接不读。

第 3 步：读 manifest，做文件级裁剪

读剩下的 manifest，拿到每个数据文件的分区值 + 每列 min/max 统计。
用查询条件做两层裁剪：
- 分区裁剪：分区值不匹配的文件跳过。
- 文件裁剪：比如 WHERE id = 123，manifest 里 id 列的 min/max 是 [100, 200]，123 在范围内才读这个文件；如果 min/max 是 [300, 400]，直接跳过这个文件。

第 4 步：读数据文件，做行级/列级裁剪

只读剩下的数据文件（Parquet）。
列裁剪：只读 SELECT 用到的列，不读全表列。
谓词下推：把 WHERE 条件下推到 Parquet reader，在 row group 级别用页统计跳过不需要的页。

第 5 步（v2 表才有）：应用删除文件

如果表有 position/equality delete 文件，在读数据时按删除标记过滤掉被删的行。

flowchart LR
    L1["Catalog"] -->|查表| L2["metadata.json<br/>current-snapshot-id"]
    L2 --> L3["manifest list"]
    L3 -->|分区范围裁剪| L4["manifest (部分)"]
    L4 -->|min/max 文件裁剪| L5["数据文件 (少量)"]
    L5 -->|列裁剪+谓词下推| L6["实际读取的行"]

4、以文件层级结构的视角，读取 iceberg 最新快照的流程

这题和 3 的区别在于「快照选择」这一步。读最新快照是默认路径；理解它之后，time travel（读历史快照）只是换一个 snapshot_id。

读最新快照的精确路径：

Catalog → 当前 metadata.json。
metadata.json 里的 current-snapshot-id 字段 → 这就是「最新快照」的 id。不需要扫快照列表找最大的，current-snapshot-id 是显式字段，指向当前生效的快照。
用这个 snapshot_id 在 snapshots 数组里找到快照记录 → 拿到 manifest list 指针。
之后同 3 第 2 步起：manifest list → manifest → 数据文件。

关键点：current-snapshot-id 是 metadata.json 里的一个字段，不是算出来的。 提交时由 writer 写死。所以「读最新快照」=「读 current-snapshot-id 指向的快照」，一步定位。

对比 time travel（读历史快照）：

time travel 不读 current-snapshot-id，而是用户指定一个 snapshot_id 或时间戳。
指定时间戳时，Iceberg 在 snapshot-log 里找「时间戳 ≤ 指定时间」的最近快照。
之后路径完全一样：那个快照的 manifest list → manifest → 数据文件。

为什么能 time travel（因果）：因为每次写都生成新快照，老快照的 manifest list 和数据文件都没被删（只是 current-snapshot-id 指向了新的）。所以任意一个老快照随时能被重新读出来——只要它的 manifest 和数据文件还在。这也是为什么 expire_snapshots 要显式设保留期：删了快照就放弃了那个时间点的表状态，不可逆。

5、以文件层级结构的视角，写入 iceberg 表何时产生了小文件？小文件是什么？有哪些？

先定义「小文件」：Iceberg 里小文件不是一个正式术语，它指远小于目标大小（默认 write.target-file-size-bytes，通常 512MB）的数据文件/元数据文件。小文件问题是 Iceberg 在对象存储上做追加写的必然产物——不是 bug，是机制代价。

何时产生小文件——按场景：

场景	为什么产生小文件	严重度
流式高频写（Flink 微批每 checkpoint 提交一次）	每个 checkpoint 都 commit 一次，每次只写几 MB 数据 → 每次一个几 MB 的 Parquet 文件。一天下来几千个小文件	最严重
分区扇出大	数据按多级分区写，每个分区只分到几行 → 每个分区一个小文件。比如按 `date + hour + city` 分区，很多组合只有几条数据	严重
小批量频繁批写	每次批写都 commit，批小 → 文件小	中
写入失败重试	写了文件但 commit 失败 → 文件落盘了但没快照引用 → 孤儿小文件	中

小文件有哪些（按层级）——这是面试官想听你分清的：

小数据文件（最常被讨论）：几 KB/几 MB 的 Parquet 文件。危害：对象存储读放大（一次查询要 open 几千个小文件，每个都有请求开销）、元数据膨胀（manifest 要记录每个文件）。
小 manifest 文件：每次写生成一个 manifest，manifest 本身也是文件。频繁写 → manifest 碎片化。危害：manifest list 越来越长，查询规划要读更多 manifest。
小 metadata 文件：每次 commit 生成一个新 metadata.json。频繁写 → metadata-log 膨胀。
小删除文件（v2 表）：流式 upsert 高频写 → 每次写都追加 equality/position delete 文件，删除文件也碎成小文件。危害：读时要合并大量删除文件，读放大严重。

为什么小文件是问题（因果链）：

对象存储每次 GET 有固定延迟（几十到几百 ms），不管文件多小。
查询要读几千个小文件 = 几千次 GET = 延迟叠加 → 查询慢。
元数据层：manifest 要记录每个小文件，manifest 变大变多 → 查询规划变慢。
这和 MapReduce 时代「大量小文件压垮 NameNode」是同源问题，只是 Iceberg 把它从 NameNode 内存压力转移到了对象存储请求放大。

6、iceberg 除了小文件还有什么文件需要治理？

Iceberg 的治理链是四步，小文件合并只是第一步。这四步对应四种不同的「垃圾」：

治理动作	存储过程	治理什么	不治理的后果
① 数据文件合并（compaction）	`rewrite_data_files`	小数据文件 → 大数据文件	查询慢（读放大）、元数据膨胀
② manifest 合并	`rewrite_manifests`	小/碎 manifest → 大 manifest	manifest list 长，查询规划慢
③ 快照过期	`expire_snapshots`	删老快照（释放 manifest list/manifest 的引用）	元数据膨胀、存储费线性涨、time travel 保留期失控
④ 孤儿文件清理	`remove_orphan_files`	物理删没有任何 manifest 引用的数据文件	失败写/compaction 残留文件占存储

逐个讲清因果：

① compaction（rewrite_data_files）

流程：读 manifest 找小文件（不 list 对象存储，靠 manifest 记录）→ 按 target-size 装箱分组（bin-packing，默认 512MB）→ 把每组小文件读出来合并写成一个大数据文件 → 生成新 manifest 指向新大文件 → 新快照原子提交。
关键：compaction 只写新文件 + 让新快照指向新文件，老小文件不删——它们仍被老快照引用。要等第 ③ 步删老快照后，老小文件才变成「无引用」。
代价：重写数据有读放大/写放大，所以按阈值触发（文件数阈值或大小阈值），不是每次写都合并。

② manifest 合并（rewrite_manifests）

频繁写导致 manifest 碎片化（每次写一个 manifest）。这个动作把多个小 manifest 合并成一个大 manifest。
危害比数据小文件隐蔽，但查询规划时要读的 manifest 数量直接影响规划延迟。

③ 快照过期（expire_snapshots）

删除早于某个时间点/早于 N 个的快照。快照被删 → 它引用的 manifest list 和 manifest 可能变成无人引用。
不直接删数据文件：一个数据文件可能被多个快照引用，只有所有引用它的快照都过期了，它才成为孤儿。这个设计是为了 time travel——删快照等于放弃那个时间点的表状态，必须显式设保留期。
代价：删了快照就不能 time travel 回那个时间点，不可逆。

④ 孤儿文件清理（remove_orphan_files）

扫描数据目录，找出不在任何 manifest 里的数据文件，物理删除。
来源：写操作失败（文件落盘了但 commit 失败）、compaction 后老小文件被新快照替代但还没被删。
安全阀：有时间阈值（默认保留最近几天修改的文件不删），避免删掉正在写、还没提交的文件——不和并发写打架。

四步的依赖关系（这是面试官想确认你理解的）：

compaction 产生新大文件，老小文件还在（被老快照引用）
  → expire_snapshots 删老快照，老小文件变成无引用孤儿
    → remove_orphan_files 物理删孤儿

顺序不能反：不删快照就直接删文件 = 删了还在被引用的数据 = 表坏了。所以 remove_orphan_files 只删「确认无引用」的文件，而「无引用」要靠 expire_snapshots 先断引用。

7、spark 的架构是什么样的？

Spark 是主从架构：一个 Driver 协调，多个 Executor 干活，ClusterManager 分配资源。

flowchart TD
    subgraph Driver["Driver (JVM 进程)"]
        D1["SparkContext<br/>(入口, 含 SparkConf)"]
        D2["DAGScheduler<br/>(切 Job→Stage)"]
        D3["TaskScheduler<br/>(切 Stage→Task, 派发)"]
        D4["SchedulerBackend<br/>(和 ClusterManager 通信)"]
    end
    subgraph CM["ClusterManager"]
        C1["YARN / K8s / Standalone Master"]
    end
    subgraph E1["Executor 1 (JVM 进程)"]
        T1["TaskSlot × N"]
        B1["BlockManager<br/>(缓存块)"]
    end
    subgraph E2["Executor 2 (JVM 进程)"]
        T2["TaskSlot × N"]
        B2["BlockManager"]
    end
    D1 --> D2 --> D3 --> D4
    D4 <--> C1
    C1 -.启动.-> E1
    C1 -.启动.-> E2
    D3 -.派发 Task.-> T1
    D3 -.派发 Task.-> T2

三个角色的职责（必须分清，面试常问）：

角色	是什么	干什么
Driver	一个 JVM 进程，跑用户 `main` 函数的地方	解析作业、构建 DAG、切分 Stage/Task、调度、跟踪任务完成、存查询元数据
Executor	JVM 进程，在工作节点上跑	执行 Task、存内存里的数据（RDD 缓存、shuffle 输出）、用 BlockManager 管缓存块
ClusterManager	外部资源管理器（YARN/K8s/Standalone Master）	给 Application 分配 Executor 资源（在哪台机器起几个 Executor）

Driver 内部三个关键组件：

SparkContext：入口，持有配置、和 ClusterManager 的连接。
DAGScheduler：把一个 Job 切成多个 Stage（按 shuffle 边界切），每个 Stage 切成多个 Task。
TaskScheduler：把 Task 派发给 Executor，跟踪完成/失败/重试。

Executor 内部：

多个 TaskSlot（任务槽，数量 = spark.executor.cores）。一个 slot 跑一个 Task。
BlockManager：管内存/磁盘上的数据块（RDD 分区、shuffle 数据、广播变量）。

一个 Application 的生命周期：

Client 提交 Application → ClusterManager 启动 Driver（或在本地起 Driver）。
Driver 向 ClusterManager 申请 Executor 资源。
ClusterManager 在工作节点起 Executor，Executor 反向注册到 Driver。
Driver 把 Task 派发给 Executor 跑。
Application 结束，Executor 释放。

8、spark 中执行一个 sql 具体是什么样的？

Spark SQL 把一条 SQL 文本变成 RDD 执行，中间经过 Catalyst 查询优化器的五个阶段。这是计算平台岗的高频题。

flowchart TD
    S["SQL 文本<br/>SELECT ..."] -->|① Parser| A["未解析逻辑计划<br/>(Unresolved Logical Plan) = AST"]
    A -->|② Analyzer| B["解析后逻辑计划<br/>(Resolved Logical Plan)"]
    B -->|③ Optimizer| C["优化后逻辑计划<br/>(Optimized Logical Plan)"]
    C -->|④ Planner| D["多个物理计划 → 选最优<br/>(Selected Physical Plan)"]
    D -->|AQE 重规划| D2["执行中重规划<br/>(shuffle map 后)"]
    D2 -->|⑤ CodeGen| E["Whole-Stage CodeGen<br/>→ RDD 执行"]
    E --> R["结果"]

逐阶段讲清：

① Parser（解析）

用 ANTLR 做词法+语法分析，把 SQL 文本变成 AST（抽象语法树）。
这一步只管语法对不对，不管 users 表存不存在、name 列在不在——AST 里的表名/列名是「未解析」的字符串。
产物叫 Unresolved Logical Plan（未解析逻辑计划）。

② Analyzer（分析）

拿着 Catalog（表/列注册表，记录所有表名、字段、类型）去校验 AST：users 表存不存在？name 列在不在 users 里？类型匹不匹配？
把「未解析」的名字绑定到真实表/列。这一步漏了，面试官会判断你只记得「有优化」没理清「先校验再优化」。
产物叫 Resolved Logical Plan（解析后逻辑计划）。

③ Optimizer（优化，Catalyst 的核心）

这里是两种优化先后作用，必须分清：
- RBO（Rule-Based Optimization，规则优化）：基于固定规则改写计划。典型：谓词下推（把 WHERE 推到数据源侧先过滤，少读数据）、列裁剪（只读用到的列）、常量折叠（1+1 算成 2）。不依赖数据分布，一定执行。
- CBO（Cost-Based Optimization，代价优化）：基于统计信息（表行数、列基数、直方图）估算多个候选计划的代价，选最便宜的。典型：join 顺序选择、join 算法选择（broadcast vs sort-merge）。
产物叫 Optimized Logical Plan（优化后逻辑计划）。

④ Planner（物理计划生成 + 选择）

SparkPlanner 把逻辑计划转成多个物理计划候选（比如同一个 join 可以用 broadcast join 也可以用 sort-merge join）。
用 Cost Model 估算每个候选的代价，选最优的那个。
产物叫 Selected Physical Plan（选定的物理计划）。

④.5 AQE（自适应查询执行，3.0+ 默认开）

AQE 不是执行前的一次性优化，而是执行中的重规划。具体在 shuffle map 阶段跑完那一刻——Spark 拿到了真实统计（每个分区实际多少行），用真实数纠正第 ④ 步的估算，重规划后面还没跑的阶段。
三个调整：① 合并小分区（coalesce shuffle partitions）；② 切换 join 策略（发现右表小，sort-merge → broadcast）；③ 倾斜处理（skew join，拆倾斜分区）。
关键限定词：执行中、shuffle map 之后、用真实统计重规划。少一个面试官都分不清你是真懂还是猜的。
版本坑：spark.sql.adaptive.enabled 在 3.2.0 从默认 false 翻成 true。.version("1.6.0") 只说开关何时引入，不说默认值何时变——这是 Spark 配置文档的常见误导，自查用 git show v3.5.9:<file>。

⑤ CodeGen（代码生成）

Whole-Stage CodeGen（全阶段代码生成）：把一个 Stage 内的多个算子融合成一段 Java 代码（而不是每个算子一个函数调用），减少虚函数调用和内存中间结果。
这是 Tungsten 的两把利器之一（另一把是堆外内存，见第 10 题）。
最终生成 RDD 的执行图，交给 DAGScheduler 切 Stage 跑。

为什么这么设计（因果）：SQL 是声明式的（说「要什么」不说「怎么做」），执行引擎要把它变成可执行的计划。Catalyst 的分层让每一步职责单一：Parser 只管语法、Analyzer 只管名字绑定、Optimizer 只管改写、Planner 只管选代价。分层 = 可插拔（自定义规则/数据源）+ 可优化（每步独立）。

9、job、stage、task、分区与并行度、rdd、transformation&action 之间的关系是什么？为什么要有 rdd？

RDD（Resilient Distributed Dataset，弹性分布式数据集）：一个只读的、分区的、容错的记录集合。四个核心属性：

分区（partitions）：数据被切成多份，每份是一个分区，分布在不同 Executor 上。分区是并行度的单位。
依赖（dependencies）：RDD 记得自己是从哪些父 RDD 来的（血缘 lineage）。
计算函数（compute）：每个分区怎么算出来（比如「对父分区每个元素 +1」）。
分区器/位置偏好（partitioner/preferredLocations）：数据怎么分桶、最好在哪台机器算（数据本地性）。

为什么要有 RDD（这是面试重点，要讲因果）：

RDD 出现前是 MapReduce。MapReduce 的痛点：每个 job 之间靠磁盘传递数据，中间结果不能复用——做一次 map-reduce 落盘，再做一次 map-reduce 又读盘。机器学习迭代算法要跑几十轮，每轮都读盘写盘，慢得不可接受。

那为什么不直接用「分布式共享内存」（DSM，让多个节点读写同一块内存）？因为 DSM 的容错代价太高：内存是易失的，一个节点挂了，它上面的数据就没了。要容错就得频繁 checkpoint（把内存状态快照到盘），开销巨大，且要同步所有写操作（保证一致性），分布式系统里极贵。

RDD 的洞察：用「限制」换「廉价容错」。

限制：RDD 是只读的，只能通过粗粒度 transformation 生成新 RDD（不能原地改某个元素）。
这个限制带来的好处：每个 RDD 都能从它的血缘（父 RDD + 计算函数）重新算出来。一个节点挂了，丢失的分区靠血缘重算即可，不需要 checkpoint 整个内存状态。
重算只算丢失的分区（粗粒度），不是整个 RDD——比 DSM 的细粒度恢复便宜得多。

一句话：RDD 是「分布式共享内存的受限版」，它用「不可变 + 血缘」换来了「廉价容错 + 内存复用」。这是 Spark 比 MapReduce 快的根因——不是「全内存所以快 100x」（那是营销话术），是「中间结果可内存复用 + 血缘容错省了落盘」。

一个 stage 可能有多个 rdd

transformation vs action

transformation（转换）：从 RDD 生成新 RDD，惰性求值（lazy）——调用时不算，只记血缘。例：map、filter、flatMap、groupByKey、join。
action（动作）：触发实际计算，把 RDD 算成结果（或写到外部）。例：collect、count、save。
为什么惰性：这样 Spark 能看到整个血缘图，做整体优化（比如谓词下推、pipeline 融合）。如果 transformation 立即执行，就没法跨算子优化。

job / stage / task / 分区的关系

flowchart TD
    A["action (如 collect)"] -->|触发| J["1 个 Job"]
    J -->|按 shuffle 边界切| S1["Stage 0"]
    J -->|按 shuffle 边界切| S2["Stage 1"]
    S2 -->|父 Stage 跑完才跑| S1
    S1 -->|每个分区一个 Task| T1["Task × N"]
    S2 -->|每个分区一个 Task| T2["Task × N"]

Job：一个 action 触发一个 Job。一个 Application 可以有多个 Job（多个 action）。
Stage：Job 按宽依赖（shuffle）切成多个 Stage。一个 Stage 内全是窄依赖，可以 pipeline 串行执行。Stage 之间靠 shuffle 传递数据。
Task：Stage 里每个分区一个 Task。Task 是 Spark 最小的执行单位，跑在一个 TaskSlot 上。
分区：RDD 的数据分片，是并行度的单位。分区数 = Task 数（一个 Stage 内）。

完整因果链：

你写一堆 transformation，它们只构建 RDD 血缘图，不执行。
调一个 action → 触发一个 Job。
DAGScheduler 扫血缘图，遇到宽依赖就切 Stage → Job 被切成多个 Stage。
每个 Stage 里每个分区生成一个 Task → TaskScheduler 派发给 Executor 跑。
Stage 之间有 shuffle 依赖：父 Stage 全跑完、shuffle 写盘，子 Stage 才能读。

宽窄依赖：

窄依赖：父 RDD 一个分区 → 子 RDD 一个分区（一对一）。例：map、filter、flatMap。不需 shuffle，可 pipeline。
宽依赖（shuffle 依赖）：父 RDD 一个分区 → 子 RDD 多个分区（一对多）。例：groupByKey、reduceByKey、join、repartition。需 shuffle。
你之前举 count 当宽依赖是错的——count 是 action 不是 transformation，不参与依赖图划分。该举 groupByKey/join/repartition。

为什么宽依赖是 Stage 边界：宽依赖要 shuffle，shuffle 要等上游全跑完写盘下游才能读——必须断开 pipeline。窄依赖不用等，可以一条龙处理完一个分区的数据。所以 DAGScheduler 用宽依赖切 Stage。

10、spark 的内存结构式什么？怎么管理的？

Spark 1.6+ 默认用 UnifiedMemoryManager（统一内存管理）。一个 Executor 的堆内存被切成几块，关键是执行内存和存储内存共享一个区域，可以互相借。

flowchart TD
    H["Executor 堆内存 (spark.executor.memory)"] --> R["预留内存 Reserved<br/>300MB 固定, Spark 内部"]
    H --> U["可用内存 Usable<br/>= (堆 - 300MB) × spark.memory.fraction(0.6)"]
    H --> P["用户内存 User Memory<br/>= 剩下的部分, 存 UDF 对象/用户数据结构"]
    U --> E["执行内存 Execution<br/>shuffle/join/agg/sort"]
    U --> S["存储内存 Storage<br/>缓存 RDD/广播变量/block"]
    E <-.可互借.-> S

逐块讲：

块	大小	存什么	能不能被挤
预留内存	固定 300MB	Spark 内部对象	不可动
用户内存	(堆-300MB) × (1-0.6) = 40%	UDF 里的对象、用户数据结构	独立，不参与统一管理
执行内存	可用内存的一部分	shuffle / join / aggregation / sort 用的内存	不能被存储挤（挤了任务就崩）
存储内存	可用内存的一部分	缓存 RDD、广播变量、block cache	能被执行挤（被挤就淘汰缓存块）

统一管理：

执行内存和存储内存共享可用内存区域，按 spark.memory.storageFraction（默认 0.5）定保底比例。
执行内存不够 → 可以挤占存储内存：把缓存的 block 淘汰掉腾地方。
反过来不行：存储内存不够时，不能挤占执行内存——执行中的数据挤了任务就崩了。
这是不对称的：执行优先级高于存储。

为什么这么设计（因果）：不同 Job 对执行/存储的需求比例是变的——shuffle 重的 Job 要执行内存，缓存多的 Job 要存储内存。固定比例会浪费。统一管理让两块按需流转，利用率高。

代价：执行内存挤占存储会导致缓存被淘汰，下游要重算——不是免费午餐。

堆外内存：

除堆内，Spark 还能用堆外内存（off-heap）：spark.memory.offHeap.enabled=true 开启，spark.memory.offHeap.size 设大小。
堆外内存是 JVM 堆之外的直接内存，不受 GC 管理（JVM 垃圾回收扫不到），适合存大量 block，避免 GC 停顿。
代价：要自己管内存释放，泄漏难排查。

Tungsten 两把利器（和内存强相关）：

堆外内存：上面讲的，绕开 GC。
全阶段 CodeGen：把多个算子融合成一段代码，减少中间对象分配——本质也是省内存和减少 GC 压力。

11、spark 调优有哪些参数？

按调什么分类记，比背参数名有效。

① 资源分配（最基础，决定并行度天花板）

参数	默认	作用
`spark.executor.instances`	-	Executor 数量
`spark.executor.cores`	-	每个 Executor 的核数（= TaskSlot 数）
`spark.executor.memory`	-	每个 Executor 堆内存
`spark.executor.memoryOverhead`	executor.memory × 0.1	堆外内存（shuffle 用堆外时）

② 内存管理（OOM 调优核心）

参数	默认	作用
`spark.memory.fraction`	0.6	堆里可用于执行+存储的比例
`spark.memory.storageFraction`	0.5	执行/存储里存储的保底比例
`spark.memory.offHeap.enabled`	false	开启堆外内存
`spark.memory.offHeap.size`	0	堆外内存大小

③ Shuffle（性能头号来源）

参数	默认	作用
`spark.sql.shuffle.partitions`	200	SQL shuffle 分区数（调大→每分区数据少→单分区内存降）
`spark.shuffle.file.buffer`	32KB	shuffle 写缓冲，调大减少磁盘小写入
`spark.shuffle.compress`	true	shuffle 输出压缩

④ 并行度

参数	默认	作用
`spark.default.parallelism`	-	RDD 算子默认分区数

⑤ AQE（3.0+，自适应）

参数	默认	作用
`spark.sql.adaptive.enabled`	3.2+ true	开 AQE
`spark.sql.adaptive.coalescePartitions.enabled`	true	合并小分区
`spark.sql.adaptive.skewJoin.enabled`	true	倾斜 join 拆分

⑥ 序列化

参数	默认	作用
`spark.serializer`	JavaSerializer	换 KryoSerializer 更快更省内存

⑦ 倾斜/广播

参数	默认	作用
`spark.sql.autoBroadcastJoinThreshold`	10MB	小表自动广播 join 阈值

set.spark.sql.broadcastTimeout = 300 默认 5 分钟，调大广播阈值那么这个也需要调大

map 端：

set.spark.sql.files.maxPartitionBytes = 512m 简单 map 可以调大，减少 maptask 数量；复杂 map 可以调小增加 maptask 数量

调优心法（不是背参数，是定位逻辑）：

慢 → 先看 Web UI：哪个 Stage 慢？Stage 里 Task 耗时分布是否倾斜？
倾斜 → AQE skew join / 加盐 / 广播小表。
OOM → 看是执行内存还是堆外：调 spark.memory.fraction、sql.shuffle.partitions、memoryOverhead。
shuffle 多 → 调 sql.shuffle.partitions、用广播 join 省 shuffle。

12、举一个 spark 调优的经验？

13、flink 的架构是怎么样的？

Flink 也是主从架构，但角色划分比 Spark 细。JobManager 是主，TaskManager 是从，ResourceManager 管资源。

flowchart TD
    subgraph JM["JobManager (主)"]
        J1["Dispatcher<br/>(接收提交, 起 JobMaster)"]
        J2["JobMaster<br/>(管一个 Job: 调度/检查点/故障恢复)"]
        J3["ResourceManager<br/>(管 TaskManager slot 资源)"]
    end
    subgraph TM1["TaskManager 1 (从, JVM 进程)"]
        S1["TaskSlot × N<br/>(资源单位)"]
        T1["Task (算子子任务)"]
    end
    subgraph TM2["TaskManager 2"]
        S2["TaskSlot × N"]
        T2["Task"]
    end
    J1 --> J2
    J2 --> J3
    J3 -.申请 slot.-> S1
    J3 -.申请 slot.-> S2
    J2 -.调度 Task.-> T1
    J2 -.调度 Task.-> T2

三个角色（必须分清，和 Spark 对比记）：

Flink 角色	对应 Spark	职责
JobManager	Driver	协调，但拆成三个子组件
- Dispatcher	-	接收 Job 提交，起 JobMaster
- JobMaster	DAGScheduler+TaskScheduler	管一个 Job 的全生命周期：调度、checkpoint 协调、故障恢复
- ResourceManager	ClusterManager	管 TaskManager 的 slot 资源（注意：是 Flink 自己的 RM，不是 YARN 的 RM）
TaskManager	Executor	跑 Task 的 JVM 进程，有多个 TaskSlot

TaskSlot（任务槽）：

TaskManager 的资源被切成多个 slot，一个 slot 跑一个 Task（算子的一个并行子任务）。
slot 数 = TaskManager 能并行跑多少 Task。
Slot Sharing（槽共享）：一个 Job 的不同算子（source / map / sink）的子任务可以共享一个 slot，让一个 slot 跑完整的一条流水线——这样 Job 的并行度不受限于单个 TaskManager 的 slot 数，且数据在 slot 内线程间传递省去网络。

术语澄清：

Flink 的 ResourceManager 和 YARN 的 ResourceManager 不是一回事。用 YARN 模式时：YARN 的 RM 给 Flink 分配 TaskManager 容器，Flink 自己的 RM 在这些容器里管 slot。两层 ResourceManager，别混。
JobManager 高可用靠 ZooKeeper（active/standby），避免单点故障。

和 Spark 架构的关键差异：

Spark 是批/微批：Job 跑完就结束。Flink 是流式常驻：Job 一直跑，数据源源不断来。
所以 Flink 多了 **Checkpoint（检查点）**机制做容错（不是血缘重算，是周期性快照状态），这是 Flink 流式容错的核心。

14、flink 中执行一个 sql 具体是什么样的？

Flink SQL 的优化器也基于 Apache Calcite（和 Spark Catalyst 不同，Spark 自研 Catalyst，Flink 用 Calcite）。流程是「SQL → 逻辑计划 → 优化 → 物理执行图 → StreamGraph → JobGraph → 执行」。

flowchart TD
    S["SQL 文本"] -->|① Calcite Parser| A["SqlNode<br/>(未校验 AST)"]
    A -->|② Validator| B["RelNode<br/>(校验后逻辑计划)"]
    B -->|③ Flink 优化器| C["优化后 RelNode<br/>(谓词/投影下推, join 重排)"]
    C -->|④ 转换| D["ExecNode → Transformation<br/>(Flink 执行节点)"]
    D -->|⑤ 生成| E["StreamGraph<br/>(流图)"]
    E -->|⑥ 生成| F["JobGraph<br/>(加 checkpoint/并行度配置)"]
    F -->|⑦ 提交| G["ExecutionGraph<br/>(JobMaster 调度)"]
    G --> H["Task 在 TaskManager 上跑"]

逐阶段讲：

① Parser（Calcite 解析）

用 Calcite 的 SQL parser 把 SQL 文本变成 SqlNode（AST）。
这一步只管语法，不管表/字段是否存在。

② Validator（校验）

用 Calcite 的 validator + Flink 的 CatalogManager（表/函数注册表）校验：表存不存在、字段在不在、类型匹不匹配。
产物是 RelNode（关系代数节点，逻辑计划）。

③ Flink 优化器（基于 Calcite 的 rule + Flink 自己的 rule）

规则优化：谓词下推（WHERE 推到 source 侧）、投影下推（只取用到的列）、常量折叠、子查询解嵌、join 重排。
Flink 特有的优化：join 算法选择（Regular/Hash/NestedLoop join）、流式 join 的 retract 处理、状态算子优化。
产物是优化后的 RelNode。

④ 转换为 ExecNode → Transformation

把逻辑 RelNode 翻译成 Flink 的 ExecNode（执行节点，每个对应一个算子如 StreamExecCalc、StreamExecJoin），再转成 Transformation（Flink 的底层算子描述）。

⑤ 生成 StreamGraph

Transformation 组装成 StreamGraph（流图，描述算子拓扑和数据流）。

⑥ 生成 JobGraph

StreamGraph 优化（算子 chain 融合，把能合并的算子合成一个 Task，省线程间通信）后变成 JobGraph。
JobGraph 里带上 checkpoint 配置、并行度、状态后端配置。

⑦ 提交执行

JobGraph 提交给 JobManager → JobMaster 生成 ExecutionGraph（并行展开后的执行图，每个算子子任务一个 Task）→ 调度到 TaskManager 跑。

版本敏感点：

Flink 1.x 用的是 Blink planner（1.9 引入，1.11 起默认）。
Flink 2.0 引入了新的 planner（重写优化器架构），流程的分层有调整。你生产环境如果是 1.x，按 Blink planner 讲；如果是 2.0，要说明用的是新 planner。具体差异建议自查对应版本文档，我不确定你生产是哪个版本。

和 Spark SQL 的关键差异：

Spark Catalyst 是自研的（基于 Scala 的规则系统）；Flink 用 Calcite（Java 生态的 SQL 框架）。
Spark 是批优化（一次优化完跑完）；Flink 是流式优化（要考虑 retract 流、状态、水位线 watermark），优化器要处理「数据会变」的情况——比如 join 的两侧都是流，结果会随两侧新数据 retract/更新。

15、举一个 flink 调优的经验？

场景：实时流任务，Kafka source → 算子做维度 lookup（查外部 MySQL/Redis 补字段）→ 聚合 → 写回。发现吞吐上不去，Web UI 显示某算子 backpressure（反压）状态 HIGH。

先讲清 backpressure 是什么（因果）：

Flink 用 credit-based 流控：下游算子告诉上游「我还有多少 buffer 额度」，上游只在下游有额度时发数据。
下游某个算子处理慢（比如外部 lookup 慢）→ 下游 buffer 满 → 不给上游 credit → 上游也被堵住 → 反压一路传到 source。
本质：流水线里最慢的算子卡住了整条链。

定位（怎么发现的）：

Flink Web UI 每个算子有 BackPressure 状态：ok（绿）/ low / high（红）。high 说明这个算子处理不过来，上游在等它。
找到第一个 high 的算子——就是瓶颈。本例是 lookup 算子（外部查询慢）。
看 Checkpoint 也能侧面验证：反压严重时 checkpoint 会变久甚至超时失败（因为算子忙不过来，barrier 对齐慢）。

解法（按代价从低到高）：

① 异步 I/O（Async I/O，最对症）

同步 lookup：一个请求等返回才能处理下一条，算子空等。
异步 lookup：发请求后不阻塞，继续发下一个，用 AsyncFunction 批量等返回。吞吐大幅提升。
代价：要保证外部系统支持并发查询（连接池够）；异步顺序保证要靠 Flink 的 AsyncWaitOrder。

② 调大瓶颈算子并行度

反压算子并行度调大（比如 4 → 16），分摊负载。
代价：要 TaskManager 有足够 slot；并行度大了状态也分得更碎，checkpoint 要管的状态分到更多 Task。

③ 调网络 buffer

taskmanager.network.memory.fraction: 0.15   # 默认 0.1, 调大给网络更多 buffer
taskmanager.network.memory.segment-size: 32KB

反压和网络 buffer 直接相关，调大网络内存能缓解 buffer 抖动。
代价：网络内存多了，堆内给状态/算子的就少了。

④ Checkpoint 调优（反压的副作用）

反压会让 checkpoint 变慢（barrier 对齐慢）。解法：
- 开 unaligned checkpoint（1.11+）：barrier 不等 in-flight 数据排空，直接越过，缓解反压下 checkpoint 超时。
- 用 RocksDB 状态后端 + 增量 checkpoint：大状态时只传增量，checkpoint 快。

env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().enableUnalignedCheckpoints();
env.getCheckpointConfig().setIncremental(true);

代价：unaligned checkpoint 要存更多 in-flight 数据，恢复时更重；RocksDB 比内存状态后端慢（读写要序列化），但能存 TB 级状态。

实际选择：先看是不是外部 lookup 慢（开 Async I/O 最对症）→ 再调并行度 → 反压导致 checkpoint 超时再开 unaligned + RocksDB 增量。

复盘要点：Flink 调优的核心是找流水线最慢的算子。反压是症状不是病因——病因永远在「某个算子处理慢」。Web UI 的 backpressure 状态 + checkpoint 耗时是定位的两手证据。