Spark 从零到进阶：数据开发工程师的系统学习指南
写在前面：如果你是一名刚接触 Spark 的数据开发工程师，面对网上零散的教程和概念感到无从下手，那么这篇文章就是为你准备的。我们将从"Spark 是什么"出发，一路走到性能调优与生产踩坑，配合大量代码示例和架构图，帮你建立完整的知识体系。本文基于 Spark 3.5.x 版本编写，并会提及 Spark 4.0 的预览方向。
目录
1. Spark 概述
1.1 什么是 Spark
Apache Spark 是一个开源的统一分析引擎，专为大规模数据处理而设计。它最初于 2009 年在加州大学伯克利分校 AMPLab 诞生，2010 年开源，2014 年成为 Apache 顶级项目。Spark 提供了 SQL、流计算、机器学习和图计算等一整套大数据处理能力，可以运行在 Hadoop YARN、Kubernetes、Standalone 等多种集群管理器上。
1.2 发展历史
| 2009 | Spark 在 UC Berkeley AMPLab 诞生 |
| 2010 | BSD 许可开源 |
| 2013 | 捐赠给 Apache 软件基金会 |
| 2014 | Apache 顶级项目；Spark 1.0 发布 |
| 2016 | Spark 2.0：DataFrame/Dataset 成为主流 API，Structured Streaming |
| 2020 | Spark 3.0：AQE、Dynamic Partition Pruning |
| 2023 | Spark 3.4 / 3.5：Spark Connect 稳定、Pandas API 完善 |
| 2024+ | Spark 4.0 预览：更好的 ANSI SQL 兼容、性能持续优化 |
1.3 Spark vs Hadoop MapReduce
很多初学者会问"Spark 会替代 Hadoop 吗？"准确地说，Spark 替代的是 MapReduce 计算引擎，而 HDFS、YARN 依然是重要的存储和资源管理组件。
| 中间结果 | 落盘（HDFS） | 优先内存，可溢写磁盘 |
| 编程模型 | Map + Reduce 两阶段 | DAG（有向无环图），多阶段 |
| 延迟 | 高（批处理） | 低（批流统一） |
| 迭代计算 | 每次迭代读写 HDFS | 内存缓存，迭代效率高 10-100x |
| 生态 | 仅 MapReduce | SQL/Streaming/ML/Graph 一体化 |
| 容错 | 基于磁盘复制 | RDD 血缘（Lineage）重算 |
1.4 核心特性
- 速度快：DAG 执行引擎 + 内存计算，比 MapReduce 快 10-100 倍
- 易用性：支持 Scala、Python、Java、R 四种语言 API，代码简洁
- 通用性：一站式覆盖批处理、SQL、流计算、ML、图计算
- 兼容性：可读写 HDFS、HBase、Cassandra、S3 等数据源，运行在 YARN/K8s/Standalone/Mesos 上
1.5 生态组件全景
- Spark Core：底层引擎，提供 RDD 抽象、任务调度、内存管理、故障恢复
- Spark SQL：结构化数据处理，DataFrame/Dataset API，兼容 Hive
- Structured Streaming：基于 Spark SQL 的流计算引擎（原 Spark Streaming 已进入维护模式）
- MLlib：分布式机器学习库
- GraphX：图计算引擎
2. Spark 架构与运行原理
2.1 核心组件
| Driver | 运行 main() 函数，创建 SparkContext，将用户程序转化为 DAG，调度 Task |
| Executor | 在 Worker 节点上启动的 JVM 进程，执行 Task 并缓存数据 |
| Cluster Manager | 集群资源管理器，负责分配 CPU/内存（YARN、K8s、Standalone） |
| Worker | 集群中可运行 Application 代码的节点 |
2.2 部署模式对比
| Local | 单机运行，多线程模拟分布式 | 开发调试、学习 |
| Standalone | Spark 自带集群管理器 | 小规模独立集群 |
| On YARN | 复用 Hadoop YARN 资源管理 | 企业最常用，与 Hadoop 生态融合 |
| On Kubernetes | 容器化部署，弹性扩缩容 | 云原生环境，近年增长迅速 |
YARN 模式又分为 yarn-cluster（Driver 运行在 AM 中，适合生产）和 yarn-client（Driver 在客户端，适合交互调试）。
2.3 Job、Stage、Task 的划分
当一个 Action 算子被触发时，Spark 会提交一个 Job。DAGScheduler 根据 RDD 的依赖关系将 Job 划分为多个 Stage，每个 Stage 内创建一组 Task，由 TaskScheduler 分发到 Executor 执行。
划分规则：
- 遇到宽依赖（Shuffle）就切分 Stage
- 一个 Stage 内的所有 Task 执行完全相同的代码，只是处理的数据分区不同
- Task 数量 = Stage 最后一个 RDD 的分区数
2.4 宽窄依赖与 Shuffle
| 窄依赖 | 父 RDD 每个分区最多被子 RDD 一个分区使用 | map、filter、union、mapPartitions |
| 宽依赖 | 父 RDD 每个分区被子 RDD 多个分区使用，需要 Shuffle | reduceByKey、groupByKey、join、distinct、repartition |
Shuffle 过程：Map 端将数据按 Key 写入磁盘缓冲区，经过排序、合并、溢写后产生 Shuffle 文件；Reduce 端通过 BlockManager 拉取属于自己的数据。Shuffle 涉及磁盘 I/O、网络 I/O 和数据序列化，是 Spark 性能瓶颈的主要来源。
2.5 统一内存模型（Spark 1.6+）
Spark Executor 的内存分为以下区域： 
- Storage Memory：缓存 RDD、Broadcast 变量
- Execution Memory：Shuffle、Join、Sort 等执行过程中的临时数据
- 两者之间可以动态借用：Execution 可以借 Storage 的空闲内存，Storage 也可以借 Execution 的，但 Execution 优先级更高，Storage 被借走的部分在需要时会被淘汰
3. 环境搭建
3.1 Local 模式（最快上手）
# 下载（以 3.5.1 为例）
wget https://archive.apache.org/dist/spark/spark-3.5.1/spark-3.5.1-bin-hadoop3.tgz
tar -xzf spark-3.5.1-bin-hadoop3.tgz
cd spark-3.5.1-bin-hadoop3
# 启动 Spark Shell（Scala）
./bin/spark-shell –master local[*]
# 启动 PySpark
./bin/pyspark –master local[*]
local[*] 表示使用所有 CPU 核心。也可以用 local[2] 指定 2 个核心。
3.2 Standalone 模式
# 1. 配置 conf/spark-env.sh
cp conf/spark-env.sh.template conf/spark-env.sh
cat >> conf/spark-env.sh << 'EOF'
export SPARK_MASTER_HOST=master
export SPARK_WORKER_CORES=4
export SPARK_WORKER_MEMORY=8g
EOF
# 2. 配置 workers
echo "worker1" > conf/workers
echo "worker2" >> conf/workers
# 3. 启动集群
./sbin/start-all.sh
# 4. 提交应用
./bin/spark-submit \\
–master spark://master:7077 \\
–class com.example.MyApp \\
myapp.jar
3.3 On YARN 模式
确保 HADOOP_CONF_DIR 或 YARN_CONF_DIR 指向 Hadoop 配置目录：
export HADOOP_CONF_DIR=/etc/hadoop/conf
# 提交到 YARN（cluster 模式）
./bin/spark-submit \\
–master yarn \\
–deploy-mode cluster \\
–driver-memory 2g \\
–executor-memory 4g \\
–executor-cores 2 \\
–num-executors 10 \\
myapp.py
3.4 Spark Shell 与 spark-submit
| spark-shell | Scala 交互式 REPL，适合探索和调试 |
| pyspark | Python 交互式 REPL |
| spark-sql | SQL 交互式命令行 |
| spark-submit | 提交应用到集群（生产环境） |
spark-submit 常用参数：
| –master | 集群管理器 URL | yarn / spark://host:7077 / local[*] |
| –deploy-mode | Driver 运行位置 | cluster / client |
| –name | 应用名称 | MySparkApp |
| –class | Java/Scala 主类 | com.example.MyApp |
| –jars | 额外依赖 JAR | –jars lib/mysql-connector.jar |
| –packages | Maven 依赖自动下载 | –packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1 |
| –files | 分发文件到工作目录 | –files config.properties |
| –conf | Spark 配置项 | –conf spark.sql.shuffle.partitions=200 |
| –driver-memory | Driver 内存 | 4g |
| –executor-memory | 每个 Executor 内存 | 8g |
| –executor-cores | 每个 Executor CPU 核数 | 4 |
| –num-executors | Executor 数量（YARN） | 20 |
| –total-executor-cores | 总核数（Standalone） | 100 |
| –archives | 分发归档文件并解压 | –archives env.tar.gz#env |
| –py-files | 额外 Python 文件/压缩包 | –py-files utils.zip |
3.5 PySpark 环境配置
# 方式一：pip 安装
pip install pyspark==3.5.1
# 方式二：使用 Spark 自带的 PySpark
export SPARK_HOME=/opt/spark
export PYTHONPATH=$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.9.7-src.zip:$PYTHONPATH
export PATH=$SPARK_HOME/bin:$PATH
# 验证
pyspark –version
在 Jupyter Notebook 中使用：
pip install jupyter
export PYSPARK_DRIVER_PYTHON=jupyter
export PYSPARK_DRIVER_PYTHON_OPTS='notebook'
pyspark –master local[*]
3.6 Spark on Kubernetes
Kubernetes（K8s）已成为大数据上云的主流部署方式。Spark on K8s 从 Spark 2.3 开始支持，Spark 3.1 进入 GA，目前在云原生场景下增长迅速。
部署模式原理：
Spark on K8s 将 Driver 和 Executor 直接作为 Pod 运行在 K8s 集群中：
- Driver Pod 启动后，通过 K8s API 动态申请 Executor Pod
- Executor Pod 运行结束后自动销毁，资源归还 K8s
- 支持动态资源分配（Dynamic Allocation），按需扩缩容
与 YARN 模式对比：
| 部署方式 | 依赖 Hadoop YARN 集群 | 容器化部署，镜像打包所有依赖 |
| 弹性扩缩容 | 依赖 YARN 队列配置，扩容较慢 | 秒级 Pod 创建，弹性能力强 |
| 资源隔离 | 队列级隔离，Container 共享 OS | Pod 级隔离，容器边界清晰 |
| 依赖管理 | 通过 –jars/–packages 分发 | 打包进 Docker 镜像，版本一致性好 |
| 生态融合 | 与 HDFS/Hive/HBase 深度集成 | 云原生生态（Prometheus/Istio/Envoy） |
| 运维复杂度 | Hadoop 运维体系成熟 | 需 K8s 运维能力，学习曲线稍陡 |
| 多租户 | YARN 队列 + Linux 用户 | Namespace + RBAC，隔离更彻底 |
| 适用场景 | 传统 Hadoop 集群、离线数仓 | 云原生、混合云、弹性计算、流批一体 |
spark-submit on K8s 命令示例：
./bin/spark-submit \\
–master k8s://https://<k8s-apiserver-host>:<port> \\
–deploy-mode cluster \\
–name spark-pi \\
–class org.apache.spark.examples.SparkPi \\
–conf spark.executor.instances=5 \\
–conf spark.executor.memory=4g \\
–conf spark.executor.cores=2 \\
–conf spark.kubernetes.container.image=registry.example.com/spark:3.5.1 \\
–conf spark.kubernetes.namespace=spark-jobs \\
–conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \\
–conf spark.kubernetes.driver.pod.name=spark-pi-driver \\
local:///opt/spark/examples/jars/spark-examples_2.12-3.5.1.jar
Docker 镜像构建要点：
# 基础镜像
FROM openjdk:11-jre-slim
# 安装 Spark
ARG SPARK_VERSION=3.5.1
ARG HADOOP_VERSION=3
RUN apt-get update && apt-get install -y curl tini && \\
curl -sL https://archive.apache.org/dist/spark/spark-${SPARK_VERSION}/spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz | tar xz -C /opt && \\
ln -s /opt/spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION} /opt/spark && \\
apt-get clean
# 安装 Python（PySpark 场景）
RUN apt-get install -y python3 python3-pip && \\
pip3 install pyspark==${SPARK_VERSION} pandas pyarrow
# 拷贝自定义 JAR 依赖（如 JDBC 驱动、数据湖 SDK）
COPY jars/mysql-connector-j-8.0.33.jar /opt/spark/jars/
COPY jars/iceberg-spark-runtime-3.5_2.12-1.5.2.jar /opt/spark/jars/
# 拷贝应用 JAR
COPY app/your-app.jar /opt/spark/user-jars/
ENTRYPOINT ["/usr/bin/tini", "–"]
构建并推送镜像：
docker build -t registry.example.com/spark:3.5.1-custom .
docker push registry.example.com/spark:3.5.1-custom
K8s 特有配置参数：
| spark.kubernetes.container.image | Docker 镜像地址 | registry/spark:3.5.1 |
| spark.kubernetes.namespace | K8s 命名空间 | spark-jobs |
| spark.kubernetes.authenticate.driver.serviceAccountName | Driver 使用的 ServiceAccount | spark |
| spark.kubernetes.driver.pod.name | Driver Pod 名称 | spark-app-driver |
| spark.kubernetes.executor.podNamePrefix | Executor Pod 名称前缀 | spark-app-exec |
| spark.kubernetes.driver.limit.cores | Driver CPU 限制 | 2 |
| spark.kubernetes.executor.limit.cores | Executor CPU 限制 | 4 |
| spark.kubernetes.driver.request.cores | Driver CPU 请求 | 1 |
| spark.kubernetes.executor.request.cores | Executor CPU 请求 | 2 |
| spark.kubernetes.memoryOverheadFactor | 堆外内存比例 | 0.2 |
| spark.kubernetes.allocation.batch.size | 每批申请 Pod 数 | 10 |
| spark.kubernetes.executor.deleteOnTermination | 完成后删除 Executor Pod | true |
| spark.kubernetes.driver.podTemplateFile | Driver Pod 模板文件 | driver-pod.yaml |
| spark.kubernetes.executor.podTemplateFile | Executor Pod 模板文件 | executor-pod.yaml |
| spark.kubernetes.file.upload.path | 应用依赖上传路径 | s3://spark-uploads/ |
| spark.kubernetes.authenticate.caCertFile | CA 证书路径 | /var/run/secrets/… |
适用场景：
- 云原生架构：与公司现有 K8s 平台统一运维，避免 Hadoop/YARN 独立集群
- 弹性扩缩容：利用 K8s Cluster Autoscaler，根据负载自动增减节点，降低成本
- 混合云/多云：同一镜像在不同云厂商 K8s 集群运行，避免厂商锁定
- 流批一体：Streaming 长任务与离线批任务共享 K8s 资源池，通过 Namespace/Quota 隔离
- 依赖隔离：不同应用使用不同镜像，彻底解决"一个集群多版本依赖冲突"问题
💡 生产建议：K8s 上建议启用 External Shuffle Service（或使用 Spark 3.0+ 的 Shuffle Data on PVC 方案），开启 Dynamic Allocation，并配置 Pod 模板以挂载共享存储（如 HDFS/S3/OSS）。
4. RDD 编程
虽然现在 Spark SQL/DataFrame 是主流 API，但理解 RDD（Resilient Distributed Dataset，弹性分布式数据集）是掌握 Spark 底层原理的关键。
4.1 RDD 五大特性
4.2 创建 RDD
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("RDDDemo").master("local[*]").getOrCreate()
sc = spark.sparkContext
# 方式一：从集合创建
rdd = sc.parallelize([1, 2, 3, 4, 5], numSlices=3)
# 方式二：从外部存储创建
rdd = sc.textFile("hdfs:///data/input.txt")
rdd = sc.wholeTextFiles("hdfs:///data/logs/") # 读取目录下所有文件
4.3 Transformation vs Action
Transformation 是懒执行的，只记录转换逻辑，不立即计算；Action 触发真正的计算。
| Transformation | map、filter、flatMap、mapPartitions、sample、union、intersection、distinct、groupByKey、reduceByKey、aggregateByKey、sortByKey、join、cogroup、cartesian、coalesce、repartition |
| Action | collect、count、take、first、takeOrdered、reduce、fold、aggregate、foreach、saveAsTextFile、countByKey、collectAsMap |
# Transformation 示例
nums = sc.parallelize([1, 2, 3, 4, 5, 6])
# map：一对一转换
squares = nums.map(lambda x: x * x) # [1, 4, 9, 16, 25, 36]
# filter：过滤
evens = nums.filter(lambda x: x % 2 == 0) # [2, 4, 6]
# flatMap：一对多展平
lines = sc.parallelize(["hello world", "hello spark"])
words = lines.flatMap(lambda line: line.split(" "))
# ["hello", "world", "hello", "spark"]
# reduceByKey：按 Key 聚合（Map 端预聚合，性能优于 groupByKey）
pairs = words.map(lambda w: (w, 1))
counts = pairs.reduceByKey(lambda a, b: a + b)
# [("hello", 2), ("world", 1), ("spark", 1)]
# join
rdd1 = sc.parallelize([("a", 1), ("b", 2)])
rdd2 = sc.parallelize([("a", "x"), ("b", "y"), ("b", "z")])
rdd1.join(rdd2).collect()
# [("a", (1, "x")), ("b", (2, "y")), ("b", (2, "z"))]
# sortByKey
sorted_counts = counts.sortByKey(ascending=True)
# Action 示例
print(counts.collect()) # 返回所有结果到 Driver
print(counts.count()) # 元素数量
print(counts.take(2)) # 取前 2 个
print(counts.first()) # 第一个
total = nums.reduce(lambda a, b: a + b) # 聚合
counts.saveAsTextFile("hdfs:///output/wordcount")
4.4 reduceByKey vs groupByKey
这是面试高频考点：
| Map 端预聚合 | ✅ 有（Combiner） | ❌ 无 |
| Shuffle 数据量 | 小 | 大 |
| 性能 | 好 | 差 |
| 适用场景 | 聚合类操作 | 需要遍历所有值的场景 |
优先使用 reduceByKey / aggregateByKey / foldByKey，它们会在 Map 端做预聚合。
4.5 持久化：cache / persist / checkpoint
from pyspark import StorageLevel
# cache() = persist(MEMORY_ONLY)
rdd.cache()
# persist 支持多种存储级别
rdd.persist(StorageLevel.MEMORY_AND_DISK) # 内存不够溢写磁盘
rdd.persist(StorageLevel.MEMORY_ONLY_SER) # 序列化后存内存，节省空间
rdd.persist(StorageLevel.MEMORY_AND_DISK_SER_2) # 序列化 + 磁盘 + 2 副本
# 解除缓存
rdd.unpersist()
# checkpoint：将 RDD 写入 HDFS 等可靠存储，切断血缘
sc.setCheckpointDir("hdfs:///checkpoint")
rdd.checkpoint()
| cache | 内存 | 保留 | 频繁重用的小数据 |
| persist | 内存/磁盘可选 | 保留 | 根据数据量选择 |
| checkpoint | HDFS | 切断 | 长血缘链、迭代计算 |
4.6 共享变量
默认情况下，Spark 会把函数中用到的变量拷贝到每个 Task，Task 对变量的修改不会回传 Driver。共享变量提供了两种特殊机制：
Broadcast Variable（广播变量）：将只读变量缓存到每个 Executor，而不是每个 Task 一份，大幅减少数据传输。
# 大数据量的查找表
lookup_table = {"US": "United States", "CN": "China", "JP": "Japan"}
bc_var = sc.broadcast(lookup_table)
rdd = sc.parallelize(["US", "CN", "JP"])
result = rdd.map(lambda code: bc_var.value.get(code, code)).collect()
# ["United States", "China", "Japan"]
Accumulator（累加器）：分布式计数器，只能在 Executor 端 add，在 Driver 端读取 value。
accum = sc.accumulator(0)
def process_line(line):
global accum
if "ERROR" in line:
accum.add(1)
return line
sc.textFile("hdfs:///logs/app.log").map(process_line).count()
print(f"Error count: {accum.value}")
⚠️ 注意：Accumulator 在 Transformation 中可能因 Task 重试导致重复计数，建议在 Action（如 foreach）中使用，或使用 Named Accumulator 并在 Driver 端确认只读取一次。
5. Spark SQL
Spark SQL 是当前 Spark 最核心、使用最广泛的模块。DataFrame/Dataset API 比 RDD 更高级、更优化。
5.1 DataFrame 与 Dataset
| 数据模型 | 无结构 | 带 Schema 的行 | 带 Schema 的强类型对象 |
| 类型安全 | 是（编译时） | 否（运行时） | 是（编译时） |
| 优化 | 无 | Catalyst | Catalyst |
| 语言 | Scala/Java/Python/R | 全部 | 主要 Scala/Java |
| Tungsten | 否 | 是 | 是 |
在 PySpark 中，DataFrame 是主要 API（Python 没有 Dataset 的编译时类型检查）。
5.2 创建 DataFrame
from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql.functions import col, sum as _sum, avg, count
spark = SparkSession.builder \\
.appName("SparkSQLDemo") \\
.master("local[*]") \\
.config("spark.sql.shuffle.partitions", "4") \\
.getOrCreate()
# 1. 从 RDD 转换
rdd = sc.parallelize([(1, "Alice", 25), (2, "Bob", 30)])
df = spark.createDataFrame(rdd, ["id", "name", "age"])
# 2. 显式指定 Schema
schema = StructType([
StructField("id", IntegerType(), False),
StructField("name", StringType(), True),
StructField("age", IntegerType(), True),
])
df = spark.createDataFrame(rdd, schema)
# 3. 读取文件
df = spark.read.csv("data/users.csv", header=True, inferSchema=True)
df = spark.read.json("data/users.json")
df = spark.read.parquet("data/users.parquet")
df = spark.read.orc("data/users.orc")
# 4. 读取 Hive 表
df = spark.sql("SELECT * FROM default.users")
# 5. JDBC
df = spark.read.format("jdbc") \\
.option("url", "jdbc:mysql://localhost:3306/mydb") \\
.option("dbtable", "users") \\
.option("user", "root") \\
.option("password", "xxx") \\
.load()
# 6. toDF
df = [(1, "Alice"), (2, "Bob")].toDF(["id", "name"]) # Scala 风格
# PySpark 中：
df = spark.createDataFrame([(1, "Alice"), (2, "Bob")], ["id", "name"])
5.3 常用 Transformation
# 选择列
df.select("name", "age").show()
df.select(col("name"), col("age") + 1).show()
# 过滤
df.filter(col("age") > 25).show()
df.where("age > 25").show()
# 分组聚合
df.groupBy("department").agg(
count("*").alias("emp_count"),
_sum("salary").alias("total_salary"),
avg("salary").alias("avg_salary")
).show()
# 排序
df.orderBy(col("age").desc()).show()
# 去重
df.select("department").distinct().show()
# 列重命名
df.withColumnRenamed("name", "employee_name")
# 新增列
df.withColumn("age_next_year", col("age") + 1)
# Join
df1.join(df2, on="id", how="inner") # inner/left/right/outer/semi/anti
df1.join(df2, df1.id == df2.emp_id, "left_outer")
# 窗口函数
from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank, dense_rank
window_spec = Window.partitionBy("department").orderBy(col("salary").desc())
df.withColumn("rank", row_number().over(window_spec)).show()
# 写数据
df.write.mode("overwrite").parquet("hdfs:///output/users")
df.write.partitionBy("department").parquet("hdfs:///output/users_by_dept")
5.4 UDF / UDAF
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, IntegerType
# 标量 UDF
def age_group(age):
if age < 25:
return "Young"
elif age < 40:
return "Mid"
else:
return "Senior"
age_group_udf = udf(age_group, StringType())
df.withColumn("age_group", age_group_udf(col("age"))).show()
# 更高效的方式：pandas UDF（向量化执行，基于 Apache Arrow）
import pandas as pd
from pyspark.sql.functions import pandas_udf
@pandas_udf(StringType())
def age_group_pandas(age: pd.Series) –> pd.Series:
return age.apply(age_group)
df.withColumn("age_group", age_group_pandas(col("age"))).show()
💡 性能提示：普通 Python UDF 每行一次 Python/JVM 序列化，性能差；Pandas UDF 以列式批量传输，性能提升 10-100 倍。Spark 3.5 还引入了 Python UDF profiling 和更高效的 Arrow 批处理。
5.5 开窗函数
from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank, dense_rank, lag, lead
window_spec = Window.partitionBy("department").orderBy(col("salary").desc())
# 排名
df.select(
"name", "department", "salary",
row_number().over(window_spec).alias("rn"),
rank().over(window_spec).alias("rank"),
dense_rank().over(window_spec).alias("dr"),
lag("salary", 1).over(window_spec).alias("prev_salary"),
lead("salary", 1).over(window_spec).alias("next_salary"),
).show()
5.6 执行计划与 Catalyst 优化器
df.explain() # 简明物理计划
df.explain(True) # 含解析/分析/优化/物理计划
df.explain("formatted") # 格式化输出（Spark 3.0+）
Catalyst 优化器是 Spark SQL 的核心，它将用户的 DataFrame/SQL 代码经过以下阶段优化：
常见优化规则：
- 谓词下推（Predicate Pushdown）：将过滤条件尽早下推到数据源
- 列裁剪（Column Pruning）：只读取需要的列
- 常量折叠（Constant Folding）：预计算常量表达式
- 布尔简化：简化 AND/OR 逻辑
5.7 Tungsten 优化
Tungsten 是 Spark 的执行引擎优化项目，目标是最大化 CPU 效率和内存效率：
5.8 AQE（Adaptive Query Execution，自适应查询执行）
Spark 3.0 引入的重大特性，在 Shuffle Map 阶段完成后根据运行时统计信息动态调整执行计划：
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.localShuffleReader.enabled", "true")
三大核心能力：
| 动态合并 Shuffle 分区 | 自动将过小的分区合并，减少 Task 数量 |
| 动态处理数据倾斜 | 自动检测倾斜分区并拆分 Join |
| 动态切换 Join 策略 | 运行时发现小表可广播时，自动从 Sort-Merge Join 切换为 Broadcast Join |
6. Spark Streaming vs Structured Streaming
6.1 Spark Streaming（DStream）
Spark Streaming 是早期的流计算方案，基于微批处理（Micro-Batch），将实时数据流按时间间隔切分为小的 RDD 批处理。 
from pyspark.streaming import StreamingContext
ssc = StreamingContext(sc, batchDuration=5) # 5 秒一个批次
lines = ssc.socketTextStream("localhost", 9999)
counts = lines.flatMap(lambda line: line.split(" ")) \\
.map(lambda w: (w, 1)) \\
.reduceByKey(lambda a, b: a + b)
counts.pprint()
ssc.start()
ssc.awaitTermination()
⚠️ Spark Streaming（DStream）从 Spark 3.4 起已标记为维护模式，新项目应使用 Structured Streaming。
6.2 Structured Streaming
Structured Streaming 是基于 Spark SQL 引擎构建的端到端流计算，将实时数据流视为一张持续追加的"无界表"。
from pyspark.sql.functions import *
# 从 Kafka 读取
df = spark.readStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "localhost:9092") \\
.option("subscribe", "orders") \\
.load()
# 解析 JSON
parsed = df.selectExpr("CAST(value AS STRING) as json") \\
.select(from_json("json", schema).alias("data")) \\
.select("data.*")
# 聚合
agg = parsed.groupBy("product_id").agg(
count("*").alias("order_count"),
sum("amount").alias("total_amount")
)
# 输出到控制台
query = agg.writeStream \\
.outputMode("complete") \\
.format("console") \\
.trigger(processingTime="10 seconds") \\
.start()
query.awaitTermination()
6.3 核心概念
Event Time（事件时间）：数据产生时自带的时间戳，而非 Spark 处理时间（Processing Time）。
Watermark（水位线）：定义了系统等待迟到数据的时间阈值，超过阈值的迟到数据将被丢弃。
# 设置 10 分钟 watermark
df_with_watermark = parsed \\
.withWatermark("event_time", "10 minutes") \\
.groupBy(
window("event_time", "5 minutes"), # 5 分钟滚动窗口
"product_id"
).agg(count("*").alias("cnt"))
Window（窗口）：
| Tumbling Window | 固定大小，不重叠（如每 5 分钟统计） |
| Sliding Window | 固定大小，可重叠（如每 1 分钟统计最近 5 分钟） |
| Session Window | 基于活动间隔动态划分（Spark 3.2+） |
Output Mode（输出模式）：
| Append | 仅输出新行 | 无聚合 |
| Complete | 输出全量结果 | 有聚合 |
| Update | 仅输出更新行 | 有聚合（Spark 2.1+） |
6.4 Source 与 Sink
| Kafka | 最常用，支持从 Kafka 读取消息 |
| File | 监听目录中新文件 |
| Socket | 测试用，从 TCP Socket 读取 |
| Rate | 测试用，每秒生成指定行数 |
| Kafka | 写入 Kafka Topic |
| File | 写入文件（Parquet/JSON/CSV） |
| Console | 控制台（调试用） |
| Foreach/ForeachBatch | 自定义写入逻辑 |
| Memory | 存储为内存表（调试用） |
6.5 Exactly-Once 语义
Structured Streaming 通过以下机制保证精确一次（Exactly-Once）：
query = agg.writeStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "localhost:9092") \\
.option("topic", "order_summary") \\
.option("checkpointLocation", "hdfs:///checkpoint/orders") \\
.outputMode("complete") \\
.start()
注意：foreachBatch 本身不保证 Exactly-Once，需要 Sink 端实现幂等或使用事务。
7. Spark 数据湖与 Lakehouse
随着大数据从"数据仓库"向"湖仓一体"演进，数据湖格式已成为 Spark 生态不可或缺的一环。本章介绍三大开源数据湖格式及 Paimon，并给出 Spark 实战代码。
7.1 数据湖概念与三剑客对比
数据湖（Data Lake）：以原始格式存储海量结构化、半结构化、非结构化数据的集中式存储，支持多种计算引擎直接读写。传统数据湖缺乏事务支持，容易产生脏数据。
Lakehouse（湖仓一体）：在数据湖存储之上增加事务层、Schema 管理、ACID 语义，兼具数据湖的灵活性和数据仓库的管理能力。
| 定位 | Databricks 主导，强绑定 Spark | 通用表格式，引擎无关 | 流式数据湖，Upsert 优先 |
| ACID 事务 | ✅ | ✅ | ✅ |
| Schema 演化 | ✅（新增/删除列） | ✅（新增/删除/重命名/调序） | ✅ |
| 分区演化 | ❌ | ✅（隐式分区演化） | ✅ |
| 时间旅行 | ✅ | ✅（Snapshot ID/时间戳） | ✅ |
| Upsert/Merge | ✅ MERGE INTO | ✅ MERGE INTO | ✅（原生 Upsert 是核心特性） |
| 删除更新 | ✅ | ✅（V2 Row-Level） | ✅（COW/MOR） |
| 流写入 | ✅ | ✅（Spark Structured Streaming） | ✅（深度流式支持） |
| 计算引擎 | Spark 为主（Delta 2.0+ 支持 Flink/Presto） | Spark/Flink/Trino/Presto/Hive | Spark/Flink/Trino/Presto |
| 社区活跃度 | Databricks 商业驱动 | Apache 顶级项目，中立开放 | Apache 顶级项目，Uber 起源 |
| 典型用户 | Databricks 客户 | Apple/Netflix/LinkedIn | Uber/Grab/字节跳动 |
| 特色 | Z-Order 优化、Delta UniForm | 隐藏分区、分区演化、快照隔离 | COW/MOR 双模式、Compaction/Clustering |
Apache Paimon 简介（原 Flink Table Store）：
Paimon 是 Flink 社区发起的流批一体存储项目，2024 年成为 Apache 顶级项目。它的核心定位是流式数据湖：
- 原生支持 Flink CDC 摄入，毫秒级延迟 Upsert
- 同时支持 Spark 读写，流批统一
- LSM Tree 架构，高吞吐写入与查询兼顾
- 适合实时数仓场景（Flink + Paimon + Spark 分析）
选型提示：如果团队以 Spark 为核心且用 Databricks，Delta Lake 最省心；如果追求引擎中立和多引擎生态，Iceberg 是趋势；如果核心需求是流式 Upsert 和增量处理，Hudi/Paimon 更合适。
7.2 Spark + Iceberg 实战
Iceberg 核心概念：
| Snapshot（快照） | 表的一次完整状态，每次写入生成新快照，旧快照保留用于时间旅行 |
| Manifest（清单） | 记录数据文件路径、分区信息、列级统计，Manifest List 管理多个 Manifest |
| Metadata File | 表元数据入口，记录当前 Snapshot、Schema、Partition Spec 等 |
| Partition Spec | 分区规则，支持隐藏分区（如按天分区但查询可按小时过滤） |
| Schema Evolution | 新增/删除/重命名/调整列顺序，无需重写数据文件 |
| Partition Evolution | 修改分区规则后旧数据不受影响，新数据按新规则写入 |
Spark 提交 Iceberg 依赖配置：
spark-submit \\
–packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2 \\
–conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \\
–conf spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog \\
–conf spark.sql.catalog.spark_catalog.type=hive \\
–conf spark.sql.catalog.local=org.apache.iceberg.spark.SparkCatalog \\
–conf spark.sql.catalog.local.type=hadoop \\
–conf spark.sql.catalog.local.warehouse=s3://my-bucket/iceberg-warehouse \\
your_app.py
建表与写入：
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, current_date, date_format
spark = SparkSession.builder.appName("IcebergDemo").getOrCreate()
# 建表
spark.sql("""
CREATE TABLE IF NOT EXISTS local.db.orders (
order_id BIGINT,
user_id BIGINT,
product_id BIGINT,
amount DECIMAL(10,2),
order_time TIMESTAMP,
dt STRING
) USING iceberg
PARTITIONED BY (dt)
""")
# 批量写入
batch_df = spark.read.parquet("s3://raw-data/orders/2024-01-15/")
batch_df.writeTo("local.db.orders").overwritePartitions()
# 流式写入（Structured Streaming）
from pyspark.sql.functions import from_json, col
kafka_schema = "order_id BIGINT, user_id BIGINT, product_id BIGINT, " \\
"amount DECIMAL(10,2), order_time TIMESTAMP, dt STRING"
stream_df = spark.readStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "localhost:9092") \\
.option("subscribe", "orders") \\
.load() \\
.selectExpr("CAST(value AS STRING) as json") \\
.select(from_json(col("json"), kafka_schema).alias("data")) \\
.select("data.*")
stream_df.writeStream \\
.format("iceberg") \\
.option("path", "local.db.orders") \\
.option("checkpointLocation", "s3://checkpoints/orders") \\
.trigger(processingTime="1 minute") \\
.start()
查询与时间旅行：
# 查询当前快照
spark.sql("SELECT * FROM local.db.orders WHERE dt = '2024-01-15'").show()
# Time Travel：按 Snapshot ID
spark.sql("""
SELECT * FROM local.db.orders.snapshotId(12345678901234567)
""").show()
# Time Travel：按时间戳
spark.sql("""
SELECT * FROM local.db.orders.timestampAsOf('2024-01-15 10:00:00')
""").show()
# 查看快照历史
spark.sql("SELECT * FROM local.db.orders.snapshots").show()
# 回滚到指定快照
spark.sql("CALL local.system.rollback_to_snapshot('db.orders', 12345678901234567)")
Schema 演化：
# 新增列
spark.sql("ALTER TABLE local.db.orders ADD COLUMN status STRING")
# 删除列
spark.sql("ALTER TABLE local.db.orders DROP COLUMN status")
# 重命名列
spark.sql("ALTER TABLE local.db.orders RENAME COLUMN amount TO total_amount")
# 调整列顺序
spark.sql("ALTER TABLE local.db.orders ALTER COLUMN order_id AFTER order_time")
分区演化：
# 从按天分区改为按月分区（旧数据不受影响）
spark.sql("ALTER TABLE local.db.orders ADD PARTITION FIELD months(order_time)")
spark.sql("ALTER TABLE local.db.orders DROP PARTITION FIELD dt")
MERGE INTO（Upsert）：
# 创建更新数据集
updates_df = spark.createDataFrame([
(1001, 2001, 3001, 99.99, "2024-01-15 14:30:00", "2024-01-15", "PAID"),
(1002, 2002, 3002, 59.99, "2024-01-15 15:00:00", "2024-01-15", "PAID"),
], ["order_id", "user_id", "product_id", "amount", "order_time", "dt", "status"])
updates_df.createOrReplaceTempView("updates")
spark.sql("""
MERGE INTO local.db.orders t
USING updates s
ON t.order_id = s.order_id
WHEN MATCHED THEN
UPDATE SET t.amount = s.amount, t.status = s.status, t.order_time = s.order_time
WHEN NOT MATCHED THEN
INSERT (order_id, user_id, product_id, amount, order_time, dt, status)
VALUES (s.order_id, s.user_id, s.product_id, s.amount, s.order_time, s.dt, s.status)
""")
7.3 Spark + Hudi 实战
Hudi 表类型对比：
| 写入方式 | 每次写入更新重写整个 Parquet 文件 | 增量写入 Log 文件，定期合并 |
| 写入延迟 | 高（重写文件） | 低（追加日志） |
| 读取延迟 | 低（直接读 Parquet） | 稍高（需合并 Parquet + Log） |
| Compaction | 不需要 | 需要（同步/异步） |
| 适用场景 | 批量更新、读多写少 | 流式 Upsert、写多读多 |
| 存储效率 | 高（列式压缩） | 稍低（Log 行式存储） |
| 典型场景 | 离线 ETL、T+1 报表 | 实时数仓、近实时分析 |
Hudi 核心概念：
| Timeline | 表的操作历史，包含 Commit/DeltaCommit/Compaction/Clean，每个 Instant 记录状态 |
| File Group | 一组文件，由 File ID 标识，包含 Base File（Parquet）和 Log File |
| File Slice | 某一时刻 File Group 的快照 = 一个 Base File + 对应 Log 文件 |
| Compaction | MOR 表将 Log 文件合并到 Base File 的过程，可同步或异步执行 |
| Clustering | 将小文件合并、按列聚簇重写，优化查询性能 |
Spark 写入 Hudi 代码示例：
from pyspark.sql import SparkSession
spark = SparkSession.builder \\
.appName("HudiDemo") \\
.config("spark.jars.packages",
"org.apache.hudi:hudi-spark3.5-bundle_2.12:0.15.0") \\
.config("spark.sql.extensions",
"org.apache.spark.sql.hudi.HoodieSparkSessionExtension") \\
.config("spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.hudi.catalog.HoodieCatalog") \\
.getOrCreate()
table_path = "s3://my-bucket/hudi/orders"
table_name = "orders"
# 批量导入（bulk_insert，最高效，不做去重）
df = spark.read.parquet("s3://raw-data/orders/")
df.write.format("hudi") \\
.option("hoodie.table.name", table_name) \\
.option("hoodie.datasource.write.recordkey.field", "order_id") \\
.option("hoodie.datasource.write.partitionpath.field", "dt") \\
.option("hoodie.datasource.write.precombine.field", "order_time") \\
.option("hoodie.datasource.write.operation", "bulk_insert") \\
.mode("overwrite") \\
.save(table_path)
# Upsert（默认操作，按主键更新或插入）
updates_df = spark.read.parquet("s3://raw-data/orders/incremental/")
updates_df.write.format("hudi") \\
.option("hoodie.table.name", table_name) \\
.option("hoodie.datasource.write.recordkey.field", "order_id") \\
.option("hoodie.datasource.write.partitionpath.field", "dt") \\
.option("hoodie.datasource.write.precombine.field", "order_time") \\
.option("hoodie.datasource.write.operation", "upsert") \\
.option("hoodie.upsert.shuffle.parallelism", 200) \\
.mode("append") \\
.save(table_path)
# MOR 表写入（流式场景推荐）
stream_df.writeStream.format("hudi") \\
.option("hoodie.table.name", table_name) \\
.option("hoodie.datasource.write.recordkey.field", "order_id") \\
.option("hoodie.datasource.write.partitionpath.field", "dt") \\
.option("hoodie.datasource.write.precombine.field", "order_time") \\
.option("hoodie.datasource.write.table.type", "MERGE_ON_READ") \\
.option("hoodie.datasource.write.operation", "upsert") \\
.option("checkpointLocation", "s3://checkpoints/hudi-orders") \\
.trigger(processingTime="1 minute") \\
.start(table_path)
Clustering / Compaction 配置：
# 异步 Clustering（写入时自动优化小文件）
hudi_options = {
"hoodie.table.name": table_name,
"hoodie.datasource.write.recordkey.field": "order_id",
"hoodie.datasource.write.partitionpath.field": "dt",
# Clustering
"hoodie.clustering.async.enabled": "true",
"hoodie.clustering.async.max.commits": "4", # 每 4 次提交触发一次
"hoodie.clustering.plan.strategy.target.file.max.bytes": "134217728", # 128MB
"hoodie.clustering.plan.strategy.small.file.limit": "629145600", # 600MB
# MOR Compaction
"hoodie.compact.inline": "true",
"hoodie.compact.inline.max.delta.commits": "5", # 每 5 次 Delta Commit 触发
"hoodie.compact.inline.trigger.strategy":
"org.apache.hudi.table.action.compact.strategy.NumCommitsAfterLastCompactionTriggerStrategy",
}
# 离线触发 Clustering
spark.sql(f"""
CALL run_clustering(
table => '{table_name}',
path => '{table_path}',
options => 'hoodie.clustering.plan.strategy.target.file.max.bytes=134217728'
)
""")
# 离线触发 Compaction
spark.sql(f"""
CALL run_compaction(
table => '{table_name}',
path => '{table_path}'
)
""")
7.4 Spark + Delta Lake
Delta Lake 核心特性：
| ACID 事务 | 基于事务日志（_delta_log），并发写入通过乐观并发控制保证一致性 |
| Time Travel | 通过版本号或时间戳读取历史数据快照 |
| Schema Enforcement | 写入时严格校验 Schema，类型不匹配直接拒绝 |
| Schema Evolution | 支持新增列、删除列（Delta 1.2+） |
| MERGE INTO | 原生 Upsert 语法，支持复杂条件 |
| OPTIMIZE + Z-ORDER | 合并小文件并按排序列聚簇，加速点查 |
| Change Data Feed | 增量读取变更数据（类似 Hudi 增量视图） |
Spark 读写 Delta 代码示例：
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder \\
.appName("DeltaDemo") \\
.config("spark.jars.packages",
"io.delta:delta-spark_2.12:3.1.0") \\
.config("spark.sql.extensions",
"io.delta.sql.DeltaSparkSessionExtension") \\
.config("spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.delta.catalog.DeltaCatalog") \\
.getOrCreate()
# 写入 Delta 表
df = spark.range(0, 1000).withColumn("value", col("id") * 10)
df.write.format("delta").mode("overwrite").save("s3://my-bucket/delta/events")
# 读取
spark.read.format("delta").load("s3://my-bucket/delta/events").show()
# SQL 建表
spark.sql("""
CREATE TABLE IF NOT EXISTS delta_events (
id LONG, value LONG
) USING delta LOCATION 's3://my-bucket/delta/events'
""")
Time Travel：
# 按版本号
df_v2 = spark.read.format("delta") \\
.option("versionAsOf", 2) \\
.load("s3://my-bucket/delta/events")
# 按时间戳
df_ts = spark.read.format("delta") \\
.option("timestampAsOf", "2024-01-15T10:30:00Z") \\
.load("s3://my-bucket/delta/events")
# SQL 语法
spark.sql("SELECT * FROM delta_events VERSION AS OF 2")
spark.sql("SELECT * FROM delta_events TIMESTAMP AS OF '2024-01-15 10:30:00'")
MERGE / OPTIMIZE / ZORDER：
# MERGE INTO（Upsert）
updates = spark.createDataFrame([(1, 999), (1001, 500)], ["id", "value"])
updates.createOrReplaceTempView("updates")
spark.sql("""
MERGE INTO delta_events t
USING updates s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
""")
# OPTIMIZE：合并小文件
spark.sql("OPTIMIZE delta_events")
# Z-ORDER：按高频过滤列聚簇排序，加速点查
spark.sql("OPTIMIZE delta_events ZORDER BY (id)")
# 清理旧快照（VACUUM，默认保留 7 天）
spark.sql("VACUUM delta_events RETAIN 168 HOURS")
7.5 选型建议
| Databricks 生态 / Spark 重度用户 | Delta Lake | 官方原生支持，Z-Order + OPTIMIZE 成熟 |
| 多引擎共享（Spark + Flink + Trino） | Iceberg | 引擎中立，社区开放，隐藏分区设计优秀 |
| 流式 Upsert / 实时数仓 | Hudi (MOR) | 原生流式支持，Compaction/Clustering 完善 |
| Flink CDC 实时入湖 + Spark 分析 | Paimon | Flink 原生流批一体，LSM 高吞吐写入 |
| 离线批处理 + 偶尔 Upsert | Iceberg / Delta | COW 模式即可满足，运维简单 |
| 严格 Schema 管理 + 审计合规 | Iceberg | 完整的 Schema/分区演化，快照隔离可审计 |
| 已投入 Hadoop 生态 / Hive 兼容 | Hudi / Iceberg | 均支持 Hive Catalog，迁移成本低 |
总结：没有"最好"的数据湖格式，只有"最合适"的选型。建议根据团队的计算引擎、更新频率、查询模式和运维能力综合决策。新项目若以 Spark 为核心且需要多引擎支持，Iceberg 是当前最安全的中立选择。
8. Spark 性能调优
性能调优是 Spark 工程师的核心能力。以下从多个维度系统讲解。
8.1 数据倾斜诊断与解决
数据倾斜是最常见的性能问题，表现为：大部分 Task 很快完成，但少数 Task 执行极慢甚至 OOM。
诊断方法：
- Spark UI 的 Stages 页面查看 Task 的 Shuffle Read/Write 数据量分布
- 观察是否有 Task 处理的数据量远超其他 Task
- 在 Spark SQL 中查看 AQE 倾斜处理日志
解决方案：
| 加盐（Salting） | 对热点 Key 加随机前缀，分散到不同 Task | 聚合类倾斜 |
| 广播 Join | 小表广播，避免 Shuffle | Join 倾斜（小表 < 100MB） |
| MapJoin / BroadcastHashJoin | AQE 自动或手动 hint | 大小表 Join |
| 拆分热点 Key | 将热点 Key 单独处理后合并 | 复杂聚合 |
| 两阶段聚合 | 先加随机前缀局部聚合，再去前缀全局聚合 | groupBy 倾斜 |
| AQE Skew Join | Spark 3.0 自动检测并拆分 | Join 倾斜（推荐） |
# 加盐示例：对热点 Key 打散
from pyspark.sql.functions import concat, lit, rand, col
# 第一步：加随机前缀
salted = df.withColumn("salted_key", concat(col("key"), lit("_"), (rand() * 10).cast("int")))
partial = salted.groupBy("salted_key").agg(sum("value").alias("partial_sum"))
# 第二步：去掉前缀再聚合
result = partial.withColumn("key", split(col("salted_key"), "_")[0]) \\
.groupBy("key").agg(sum("partial_sum").alias("total"))
# 广播 Join 提示
from pyspark.sql.functions import broadcast
df_large.join(broadcast(df_small), "key")
# SQL Hint
spark.sql("SELECT /*+ BROADCAST(s) */ * FROM large l JOIN small s ON l.key = s.key")
8.2 Shuffle 调优
# 1. 调整 Shuffle 分区数（默认 200，通常需调整）
spark.conf.set("spark.sql.shuffle.partitions", "200") # 通用建议：总核数的 2-3 倍
# 2. 启用 AQE 自动合并分区
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
# 3. Shuffle 压缩
spark.conf.set("spark.shuffle.compress", "true")
spark.conf.set("spark.shuffle.spill.compress", "true")
spark.conf.set("spark.io.compression.codec", "zstd") # zstd 比 snappy 压缩率更好
# 4. Shuffle 文件缓冲区
spark.conf.set("spark.shuffle.file.buffer", "1MB") # 默认 32KB
spark.conf.set("spark.reducer.maxSizeInFlight", "96MB") # 默认 48MB
# 5. RDD API 中使用 coalesce 减少分区
small_df = large_df.coalesce(10) # 窄依赖，不 Shuffle
# repartition 会 Shuffle
repartitioned = df.repartition(200, col("date"))
8.3 内存调优
# spark-submit 内存参数
–driver-memory 4g \\
–executor-memory 8g \\
–executor-cores 4 \\
–num-executors 20 \\
–conf spark.memory.fraction=0.6 \\ # Spark 内存占比（默认 0.6）
–conf spark.memory.storageFraction=0.5 # Storage 占 Spark 内存比例（默认 0.5）
Executor 内存规划原则：
- 每个 Executor 建议 4-8GB，过大易导致 GC 压力
- 每个 Executor 2-5 个 Core，过多 Core 导致 HDFS I/O 竞争
- 预留系统内存：spark.memory.fraction 默认为 0.6，即 60% 给 Spark，40% 给用户和其他开销
序列化选择：
| Java Serialization（默认） | 兼容好 | 慢、体积大 |
| Kryo Serialization | 快 10x、体积小 | 需注册类 |
spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
# Scala: spark.registerKryoClasses(Array(classOf[MyClass]))
缓存策略：
- 优先使用 MEMORY_AND_DISK_SER 而非 MEMORY_ONLY，避免 OOM
- DataFrame 建议用 spark.catalog.cacheTable() 或 df.cache()，列式存储更高效
- 及时 unpersist() 不再使用的缓存
8.4 并行度调优
- 合理设置分区数：分区太少 → 并行度不足；分区太多 → Task 调度开销大
- 经验值：分区数 ≈ 总 Executor Core 数的 2-3 倍
- 读取时控制并行度：
- HDFS 文件的 InputSplit 数量
- spark.sql.files.maxPartitionBytes（默认 128MB）
- AQE 自动调整：Spark 3.0+ 优先使用 AQE
8.5 数据本地性
Spark 倾向于将 Task 调度到数据所在节点，减少网络传输：
| PROCESS_LOCAL | 数据在同一 JVM 中（最快） |
| NODE_LOCAL | 数据在同一节点 |
| RACK_LOCAL | 数据在同一机架 |
| ANY | 数据在任意位置（最慢） |
通常不需要手动调整，但如果集群网络延迟高，可适当增大 spark.locality.wait（默认 3s）。
8.6 小文件问题
问题：大量小文件导致 Task 数量过多，调度开销巨大。
# 1. 写入时合并小文件
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")
# 2. 读取时合并小文件
df = spark.read.option("mergeSchema", "true").parquet("hdfs:///data/")
# 3. 使用 Hive 的 CombineHiveInputFormat（对 Hive 表）
spark.conf.set("spark.hadoop.hive.input.format",
"org.apache.hadoop.hive.ql.io.CombineHiveInputFormat")
# 4. 写入后用 DISTRIBUTE BY 控制输出文件数
df.write.partitionBy("dt").saveAsTable("table")
# 或在 SQL 中
# INSERT OVERWRITE TABLE t PARTITION(dt) SELECT … DISTRIBUTE BY dt
8.7 列式存储与数据格式
| Text/CSV | 行式 | 无/弱 | 通用但慢 |
| JSON | 行式 | 无 | 解析开销大 |
| Parquet | 列式 | Snappy/Gzip/Zstd | Spark 默认推荐，列裁剪高效 |
| ORC | 列式 | Zlib/Snappy/Zstd | Hive 生态好，ACID 支持 |
| Avro | 行式 | Snappy | Schema 演进好，Kafka 常用 |
# 写入 Parquet（推荐格式）
df.write.mode("overwrite") \\
.option("compression", "zstd") \\
.parquet("hdfs:///data/output")
# 分区裁剪 + 谓词下推（Parquet/ORC 自动支持）
spark.sql("SELECT name, age FROM users WHERE dt = '2024-01-01' AND age > 25")
# 只读 dt=2024-01-01 分区，且只扫描 name、age 列
8.8 Join 策略选择
Spark 支持多种 Join 实现，理解其原理对性能调优至关重要：
| Broadcast Hash Join (BHJ) | 小表 < spark.sql.autoBroadcastJoinThreshold（默认 10MB） | 无 Shuffle，最快 |
| Sort-Merge Join (SMJ) | 大表 Join 大表（默认） | 需 Shuffle + 排序，稳定但慢 |
| Shuffled Hash Join (SHJ) | 配置开启且大小表均适合构建哈希表 | Shuffle 但不排序，比 SMJ 快 |
| Broadcast Nested Loop Join (BNLJ) | 无等值条件的 Join（如 CROSS JOIN） | 无 Shuffle，但 O(n×m) |
| Cartesian Product | 无条件 Join | 结果集可能极大，慎用 |
# 手动指定 Join 策略 Hint
df1.join(broadcast(df2), "key") # 推荐：广播小表
spark.sql("SELECT /*+ MERGE(l, r) */ * FROM large l JOIN large r ON l.id = r.id")
spark.sql("SELECT /*+ SHUFFLE_HASH(l, r) */ * FROM large l JOIN medium r ON l.id = r.id")
调优建议：
- 大小表 Join：优先 Broadcast Hash Join，可适当调大 autoBroadcastJoinThreshold（如 100MB）
- 大表 Join 大表：确保 Join Key 数据类型一致，避免隐式转换导致 Shuffle
- 多表 Join：注意 Join 顺序，小表尽量先参与
- AQE 开启后，运行时自动切换为 Broadcast Join
8.9 分区策略
合理的分区设计是性能优化的基础：
# 写入时按字段分区（目录分区）
df.write.partitionBy("dt", "region").parquet("hdfs:///data/sales")
# 生成目录结构：/data/sales/dt=2024-01-01/region=CN/part-xxx.parquet
# 分桶（Bucket）：相同 Key 的数据落在同一文件，避免 Join 时 Shuffle
df.write.bucketBy(32, "user_id").sortBy("user_id").saveAsTable("bucketed_users")
# 读取时分区裁剪自动生效
spark.sql("SELECT * FROM sales WHERE dt = '2024-01-01'") # 只读一个分区目录
| 分区（Partition） | 高基数的查询过滤字段（如日期） | 分区数不宜过多（< 10000） |
| 分桶（Bucket） | 频繁 Join 或聚合的字段 | 桶数需为 2 的幂，与 Join 表一致 |
| 聚簇（Clustering） | 数据湖中自动优化文件布局 | Iceberg/Hudi/Delta Lake 支持 |
8.10 Spark UI 实战调试指南
Spark UI（默认端口 4040）是性能调优最重要的工具，能直观展示作业执行细节。以下逐页讲解排查思路。
Jobs 页面：
- 展示所有 Job 的状态（Succeeded/Failed/Running）、耗时、Stage 数
- 点击 Job 进入详情，查看 DAG 可视化图：每个方框代表一个 Stage，箭头表示 Shuffle 依赖
- 任务耗时分布：关注 Duration 列，若某 Job 耗时远超预期，进入对应 Stage 排查
- 失败任务定位：Failed Jobs 区域直接显示异常堆栈，点击 Failed Stage 查看 Task 失败日志
Stages 页面（核心！）：
- 展示每个 Stage 的 Task 数、输入/输出数据量、Shuffle Read/Write
- Shuffle Read 列：关注 Min / 25th / Median / 75th / Max，若 Max 远大于 Median（如 Max 是 Median 的 5 倍以上），说明数据倾斜
- Shuffle Write 列：Map 端写出的数据量，倾斜同样会表现为分布不均
- Task Time 分布：GC Time 占比过高说明内存压力大；Scheduler Delay 过高说明资源不足
- 点击 Stage 详情可查看每个 Task 的指标，支持排序
倾斜识别要点：在 Stages 页面查看 Summary Metrics 表，比较 Task Max 与 Median 的 Shuffle Read Size。如果 Max 是 Median 的数倍甚至数十倍，基本可以确认倾斜。
Storage 页面：
- 展示缓存的 RDD/DataFrame，包括存储级别（Memory/Deserialized 等）、缓存分区数、占用内存/磁盘大小
- 缓存命中率：Fraction Cached 列显示已缓存分区比例，若偏低说明部分分区未缓存（Executor 丢失或内存不足被淘汰）
- 分区大小：Size in Memory/ExternalBlockStore 列可判断分区是否均匀
- 内存 vs 磁盘：若大量数据在 Disk 上，说明内存不足，需调整缓存策略或增大内存
Environment 页面：
- 展示所有 Spark 配置属性（System Properties、Classpath、Spark Properties）
- 配置排查：确认提交参数是否生效（如 spark.sql.shuffle.partitions、spark.executor.memory）
- 可对比运行配置与预期配置，排查"参数没生效"类问题
Executors 页面：
| Task Time (Total/GC) | GC Time 占 Total 10%+ 说明内存压力大，需调内存或减少缓存 |
| Shuffle Read/Write | 各 Executor 的 Shuffle 数据量，差异大说明倾斜 |
| Input Size/Records | 读取数据量分布，不均匀可能是文件大小不均 |
| Storage Memory | 已用/总缓存内存，满了会溢写磁盘 |
| Failed Tasks | 非零需查看日志，可能 OOM 或数据异常 |
| Log 链接 | 点击 stdout/stderr 查看 Executor 日志（YARN 模式需通过 RM 代理） |
SQL 页面（Spark SQL / DataFrame 作业必看）：
- 展示 SQL 查询的完整执行计划 DAG，节点显示 Scan/Filter/Project/Join/Aggregate/Sort 等算子
- 点击节点查看详细指标：行数、数据量、耗时、CodeGen 信息
- Join 识别：节点显示 SortMergeJoin/BroadcastHashJoin/ShuffledHashJoin，确认是否走了预期的 Join 策略
- Broadcast 识别：若看到 BroadcastExchange 节点，说明小表被广播；若大表被广播可能 OOM
- AQE Skew Partition：开启 AQE 后，倾斜 Join 会显示 CustomShuffleReader 节点，标注拆分的分区数
- Scan 节点：查看 number of files read、size of files read、partition filters、data filters，确认分区裁剪和谓词下推是否生效
Structured Streaming 页面：
- Input Rate：每秒输入数据条数，反映流量波动
- Processing Rate：每秒处理数据条数，若持续低于 Input Rate 说明处理能力不足
- Batch Duration：每个微批的处理耗时，若持续接近 Trigger Interval 说明背压严重
- Watermark 信息：展示当前 Watermark 位置、迟到数据丢弃情况
- State Operator：状态算子（聚合/Join）的状态行数、内存占用
实战：通过 UI 定位数据倾斜的步骤：
8.11 资源规划与配置参数
Executor 资源规划公式：
推荐配置：
–num-executors = ceil(总数据量 / (executor-cores × 单核算力))
–executor-cores = 3 ~ 5（建议 4）
–executor-memory = 4g ~ 8g（建议 6g）
–driver-memory = 2g ~ 4g（复杂作业建议 4g+）
经验法则：
1. 每个 Executor 分配 4 Core，6GB 内存为黄金配比
2. YARN 总核数 = num-executors × executor-cores
3. 预留 10%~20% 集群资源给系统和其他服务
4. Executor 数量 = 集群总核数 / executor-cores × 0.8
常见踩坑：
| executor-memory 过大导致 GC | 单 Executor 超过 8GB，JVM GC 停顿时间长 | 控制在 4-8GB，增加 Executor 数量而非单 Executor 内存 |
| cores 过多导致 HDFS 并发 | 单 Executor 超过 5 Core，HDFS 客户端并发写导致超时 | 控制在 3-5 Core，HDFS 并发度 = executor-cores × num-executors |
| Driver OOM | collect() 大量数据或广播大表 | 增大 driver-memory；避免 collect 大数据集 |
| YARN Container 被 Kill | 物理内存超 Container 限制 | 增大 spark.yarn.executor.memoryOverhead |
YARN 模式内存开销：
YARN Container 内存 = spark.executor.memory + spark.executor.memoryOverhead
memoryOverhead 默认值 = max(executor-memory × spark.kubernetes.memoryOverheadFactor, 384MB)
YARN 模式默认 factor = 0.10
示例：
–executor-memory 6g
默认 memoryOverhead = max(6g × 0.10, 384MB) ≈ 614MB
Container 总内存 = 6g + 614MB ≈ 6.7GB
调优建议：
–conf spark.yarn.executor.memoryOverhead=1g # 显式指定，避免默认不足
关键配置参数速查表：
| spark.sql.shuffle.partitions | 200 | Shuffle 后分区数 | 总核数 × 2-3（开启 AQE 后可适当调大） |
| spark.executor.memory | 1g | 每个 Executor 堆内存 | 4g-8g |
| spark.executor.cores | 1 | 每个 Executor CPU 核数 | 4 |
| spark.executor.instances | — | Executor 数量（YARN/K8s） | 按集群资源计算 |
| spark.sql.adaptive.enabled | false（3.x 建议 true） | 开启 AQE 自适应执行 | true |
| spark.sql.adaptive.coalescePartitions.enabled | true | AQE 自动合并小分区 | true |
| spark.sql.adaptive.coalescePartitions.minPartitionNum | 1 | 合并后最小分区数 | 总核数 × 1 |
| spark.sql.adaptive.advisoryPartitionSizeInBytes | 64MB | AQE 目标分区大小 | 128MB |
| spark.sql.adaptive.skewJoin.enabled | true | AQE 自动处理倾斜 Join | true |
| spark.sql.adaptive.skewJoin.skewedPartitionFactor | 5 | 倾斜分区判定因子（倍数） | 5（默认即可） |
| spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes | 256MB | 倾斜分区最小数据量阈值 | 256MB |
| spark.sql.autoBroadcastJoinThreshold | 10MB | 小表自动广播阈值 | 100MB（视内存调整） |
| spark.serializer | org.apache.spark.serializer.JavaSerializer | 序列化器 | org.apache.spark.serializer.KryoSerializer |
| spark.memory.fraction | 0.6 | Spark 内存占 Executor 比例 | 0.6（默认，缓存多可降至 0.5） |
| spark.memory.storageFraction | 0.5 | Storage 占 Spark 内存比例 | 0.5（默认） |
| spark.speculation | false | 推测执行（慢 Task 备份） | true（集群空闲时开启） |
| spark.sql.files.maxPartitionBytes | 128MB | 读取文件时单个分区最大字节 | 128MB（大集群可调至 256MB） |
| spark.sql.files.openCostInBytes | 4MB | 打开文件的估算开销（小文件合并用） | 8MB（小文件多时增大） |
| spark.sql.broadcastTimeout | 300s | 广播等待超时 | 600（大表广播时） |
| spark.network.timeout | 120s | 网络通信超时 | 300s（大 Shuffle 时） |
| spark.streaming.kafka.consumer.cache.enabled | true | 缓存 Kafka 消费者 | true（默认，避免重复创建） |
| spark.sql.crossJoin.enabled | false | 允许笛卡尔积 | false（默认，禁止意外笛卡尔积） |
| spark.shuffle.service.enabled | false | External Shuffle Service | true（YARN 动态分配必须） |
| spark.dynamicAllocation.enabled | false | 动态资源分配 | true（配合 ESS） |
| spark.sql.session.timeZone | JVM 默认 | 会话时区 | Asia/Shanghai |
9. Spark 常见问题与踩坑
9.1 OOM（OutOfMemoryError）
| Driver OOM | collect() 拉取大量数据到 Driver | 用 take()/sample()/write 替代；增大 driver-memory |
| Executor OOM | 数据倾斜、缓存过多、大表广播 | 解决倾斜；调整缓存策略；增大 executor-memory |
| GC Overhead | JVM 堆外内存不足 | 增大 spark.executor.memoryOverhead（默认 executor-memory × 0.1） |
| Python Worker OOM | Pandas UDF 内存占用大 | 增大 Arrow 批量；分批处理 |
# 堆外内存配置（重要！）
–conf spark.executor.memoryOverhead=2g \\
–conf spark.driver.memoryOverhead=1g \\
–conf spark.python.worker.memory=2g
9.2 数据倾斜
见 8.1 节。典型症状：99% 的 Task 很快完成，1% 的 Task 卡住不动。
9.3 Shuffle Fetch Failed
org.apache.spark.shuffle.FetchFailedException:
Failed to connect to /xxx:xxxx
常见原因与解决：
- Executor 内存不足导致被 YARN Kill：增大 executor-memory / memoryOverhead
- 网络超时：增大 spark.shuffle.io.maxRetries（默认 3）和 spark.shuffle.io.retryWait（默认 5s）
- Shuffle 文件过大：增加 Shuffle 分区数
- 节点故障：开启 spark.shuffle.service.enabled=true（External Shuffle Service）
9.4 序列化错误
org.apache.spark.SparkException: Task not serializable
原因：在 RDD/DataFrame 的闭包中引用了不可序列化的对象（如数据库连接、非序列化的外部类）。
解决：
- 在函数内部创建不可序列化对象（如数据库连接），不要在 Driver 端创建后传入
- 使用 foreachPartition / mapPartitions，每个分区创建一次连接
- 使用 Kryo 序列化并注册类
# ✅ 正确：在分区内创建连接
def process_partition(rows):
conn = create_db_connection() # 每个分区创建一次
for row in rows:
conn.insert(row)
conn.close()
df.foreachPartition(process_partition)
9.5 时区问题
# Spark 默认使用 UTC 时区，可能导致时间差 8 小时
spark.conf.set("spark.sql.session.timeZone", "Asia/Shanghai")
# 读取时指定时区
df = spark.read.option("timestampFormat", "yyyy-MM-dd HH:mm:ss") \\
.option("timeZone", "Asia/Shanghai").csv("data.csv")
9.6 UDF 性能问题
- Python UDF 性能差，优先使用 Spark SQL 内置函数
- 必须用 UDF 时，使用 Pandas UDF（向量化）
- 避免在 UDF 中创建重型对象，用 mapPartitions 替代
9.7 其他常见坑
| count() 触发重算 | 未 cache 的 RDD/DataFrame，每次 Action 都从头计算 |
| collect() 内存溢出 | 确保结果集不超过 Driver 内存 |
| 隐式类型转换错误 | 数字与字符串比较时注意类型 |
| Hive 分区不生效 | 执行 MSCK REPAIR TABLE 或添加分区 |
| Spark UI 看不到 | YARN cluster 模式下通过 ResourceManager 代理访问 |
| 文件已存在报错 | 使用 .mode("overwrite") 或先删除 |
10. 端到端实战项目
10.1 项目背景：电商用户行为日志分析平台
项目目标：构建一个流批一体的电商用户行为分析平台，实时采集用户点击流数据，结合业务库数据，完成实时指标统计和离线报表分析。
数据源：
| Kafka 用户点击流 | 页面浏览、点击、搜索、加购等行为事件 | Flume/SDK → Kafka |
| MySQL 订单库 | 订单主表、订单明细 | Flink CDC / Spark JDBC 全量+增量 |
| MySQL 用户库 | 用户基本信息、地域信息 | 每日全量同步 |
| MySQL 商品库 | 商品分类、价格、品牌 | 每日全量同步 |
技术栈：
- 计算引擎：Spark 3.5（Structured Streaming 实时 + Spark SQL 离线）
- 数据湖：Apache Iceberg（ODS/DWD/DWS/ADS 分层存储）
- 消息队列：Kafka
- 缓存：Redis（实时指标查询）
- 业务库：MySQL（ADS 报表数据）
- 资源调度：YARN / K8s
- 监控：Prometheus + Grafana
分层架构： 
10.2 项目架构图
10.3 核心代码
1. Kafka 消费 + 数据清洗写入 Iceberg ODS：
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, current_timestamp
from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType
spark = SparkSession.builder \\
.appName("UserBehaviorODS") \\
.config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog") \\
.config("spark.sql.catalog.local.type", "hadoop") \\
.config("spark.sql.catalog.local.warehouse", "s3://lakehouse/warehouse") \\
.getOrCreate()
schema = StructType([
StructField("user_id", LongType()),
StructField("event_type", StringType()), # view/click/cart/search/buy
StructField("product_id", LongType()),
StructField("category_id", LongType()),
StructField("event_time", StringType()),
StructField("device", StringType()),
StructField("ip", StringType()),
])
kafka_df = spark.readStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092") \\
.option("subscribe", "user_behavior") \\
.option("startingOffsets", "latest") \\
.option("maxOffsetsPerTrigger", 100000) \\
.load()
parsed = kafka_df.select(
from_json(col("value").cast("string"), schema).alias("data")
).select("data.*") \\
.withColumn("ingest_time", current_timestamp()) \\
.withColumn("dt", col("event_time").substr(1, 10))
# 写入 Iceberg ODS（追加模式）
parsed.writeStream \\
.format("iceberg") \\
.option("path", "local.ods.user_behavior") \\
.option("checkpointLocation", "s3://checkpoints/ods_user_behavior") \\
.trigger(processingTime="30 seconds") \\
.outputMode("append") \\
.start()
2. 维度 Join（Broadcast）写入 DWD：
from pyspark.sql.functions import broadcast
# 加载维度表（小表广播）
dim_user = spark.read.format("iceberg").load("local.dim.dim_user")
dim_product = spark.read.format("iceberg").load("local.dim.dim_product")
# 从 ODS 流式读取
ods_stream = spark.readStream \\
.format("iceberg") \\
.option("stream-from-timestamp", str(int((__import__('time').time() – 86400) * 1000))) \\
.load("local.ods.user_behavior")
# 维度关联：广播小表避免 Shuffle
dwd = ods_stream.alias("o") \\
.join(broadcast(dim_user).alias("u"),
col("o.user_id") == col("u.user_id"), "left") \\
.join(broadcast(dim_product).alias("p"),
col("o.product_id") == col("p.product_id"), "left") \\
.select(
col("o.user_id"),
col("u.age_group"),
col("u.city"),
col("o.event_type"),
col("o.product_id"),
col("p.category_name"),
col("p.brand"),
col("p.price"),
col("o.event_time"),
col("o.dt"),
)
# Upsert 写入 DWD
dwd.writeStream \\
.format("iceberg") \\
.option("path", "local.dwd.dwd_user_behavior") \\
.option("checkpointLocation", "s3://checkpoints/dwd_user_behavior") \\
.trigger(processingTime="1 minute") \\
.start()
3. 窗口聚合写入 DWS：
from pyspark.sql.functions import window, count, sum as _sum, when, col
# 实时窗口聚合：每 1 分钟滚动窗口，统计各分类 PV/UV/加购数
dws_agg = dwd \\
.withWatermark("event_time", "10 minutes") \\
.groupBy(
window(col("event_time"), "1 minute"),
col("category_name"),
col("dt"),
).agg(
count("*").alias("pv"),
count("user_id").alias("event_count"),
_sum(when(col("event_type") == "cart", 1).otherwise(0)).alias("cart_count"),
_sum(when(col("event_type") == "buy", col("price")).otherwise(0)).alias("gmv"),
).select(
col("window.start").alias("window_start"),
col("window.end").alias("window_end"),
col("category_name"),
col("pv"),
col("event_count"),
col("cart_count"),
col("gmv"),
col("dt"),
)
dws_agg.writeStream \\
.format("iceberg") \\
.option("path", "local.dws.dws_category_realtime") \\
.option("checkpointLocation", "s3://checkpoints/dws_category_rt") \\
.outputMode("append") \\
.trigger(processingTime="1 minute") \\
.start()
4. ADS 结果输出到 MySQL/Redis：
def write_to_mysql(batch_df, batch_id):
"""将每批次结果写入 MySQL（幂等：REPLACE INTO）"""
batch_df.write \\
.format("jdbc") \\
.option("url", "jdbc:mysql://mysql:3306/ads") \\
.option("dbtable", "ads_category_realtime") \\
.option("user", "spark") \\
.option("password", "xxx") \\
.option("driver", "com.mysql.cj.jdbc.Driver") \\
.mode("append") \\
.save()
def write_to_redis(batch_df, batch_id):
"""将实时指标写入 Redis，供大屏查询"""
import redis
r = redis.Redis(host='redis', port=6379, db=0)
for row in batch_df.collect():
key = f"ads:category:{row['category_name']}:{row['window_start']}"
r.hset(key, mapping={
"pv": row["pv"],
"cart_count": row["cart_count"],
"gmv": float(row["gmv"]),
})
r.expire(key, 86400)
# 双流写入
dws_agg.writeStream \\
.foreachBatch(lambda df, id: (write_to_mysql(df, id), write_to_redis(df, id))) \\
.option("checkpointLocation", "s3://checkpoints/ads_sink") \\
.trigger(processingTime="1 minute") \\
.start()
5. 离线补数（批式回刷历史数据）：
from datetime import datetime, timedelta
def backfill(start_date, end_date):
"""批量回刷指定日期范围的数据"""
spark = SparkSession.builder.appName("Backfill").getOrCreate()
current = datetime.strptime(start_date, "%Y-%m-%d")
end = datetime.strptime(end_date, "%Y-%m-%d")
while current <= end:
dt = current.strftime("%Y-%m-%d")
print(f"Backfilling {dt} …")
# 读取 ODS 指定分区
ods_df = spark.read.format("iceberg") \\
.load("local.ods.user_behavior") \\
.where(col("dt") == dt)
# 关联维度
dwd_df = ods_df.alias("o") \\
.join(broadcast(dim_user), "user_id", "left") \\
.join(broadcast(dim_product), "product_id", "left")
# 覆盖写入 DWD 指定分区
dwd_df.writeTo("local.dwd.dwd_user_behavior") \\
.overwritePartitions()
current += timedelta(days=1)
spark.stop()
# 执行回刷
backfill("2024-01-01", "2024-01-15")
10.4 项目要点
幂等写入与 Checkpoint 管理：
- Checkpoint 目录存储 Kafka offset、聚合状态和事务元数据，删除 Checkpoint = 从头消费
- 生产环境 Checkpoint 应放在可靠存储（HDFS/S3），并设置生命周期管理
- 幂等写入策略：Iceberg 通过 MERGE INTO 按主键 Upsert；MySQL 使用 REPLACE INTO 或 INSERT … ON DUPLICATE KEY UPDATE
- 重新部署时保留 Checkpoint，仅在需要重置时手动删除
延迟数据处理（Watermark + allowedLateness）：
# 设置 10 分钟 Watermark：允许事件时间最多比处理时间晚 10 分钟
df.withWatermark("event_time", "10 minutes")
# Iceberg 流式写入可配合 to-snapshot 处理迟到数据
# 超过 Watermark 的数据会被丢弃，但 Iceberg 的 Time Travel 可追溯
# 对于重要的迟到数据，可在 ODS 层保留全量，通过离线补数修复 DWS/DWS
- Watermark 阈值根据业务延迟特征设置（一般 5-30 分钟）
- 对于严重延迟的数据（如客户端断网数小时），建议通过离线批处理补数修正
- 在 ODS 层保留原始数据（Append Only，不删不改），作为"真相源"
监控告警（Streaming Query Listener）：
from pyspark.sql.streaming import StreamingQueryListener
class MyListener(StreamingQueryListener):
def onQueryStarted(self, event):
print(f"Query started: {event.id}")
def onQueryProgress(self, event):
progress = event.progress
print(f"Batch {progress.batchId}: "
f"input={progress.numInputRows} rows, "
f"rate={progress.inputRowsPerSecond:.1f}/s, "
f"duration={progress.batchDuration}ms")
# 告警条件：处理速率低于输入速率（消费滞后）
if progress.inputRowsPerSecond > progress.processedRowsPerSecond * 1.5:
send_alert(f"Consumer lag! input={progress.inputRowsPerSecond}, "
f"processed={progress.processedRowsPerSecond}")
# 告警：批次耗时异常
if progress.batchDuration > 120000: # 超过 2 分钟
send_alert(f"Batch duration too long: {progress.batchDuration}ms")
def onQueryTerminated(self, event):
print(f"Query terminated: {event.id}, exception={event.exception}")
if event.exception:
send_alert(f"Streaming query failed: {event.exception}")
spark.streams.addListener(MyListener())
def send_alert(msg):
"""对接 Prometheus AlertManager / 钉钉 / 企业微信"""
import requests
requests.post("https://alert-webhook.example.com/", json={"text": msg})
常见问题：
| 小文件过多 | 每批次产生大量小 Parquet 文件 | 开启 Iceberg write.distribution-mode=hash；定期执行 rewrite_da
（课程设计）基于java+springboot+vue前后端分离摩托车销售商城系统设计与实现-计算机毕设 附源码 91694
基于java+springboot+vue前后端分离摩托车销售商城系统设计与实现摘 要
本文设计并实现了一个基于Java、Spring Boot与Vue的前后端分离摩托车销售商城平台。后端采用Spring Boot构建RESTful服务，负责用户管理、商品信息维护、订单处理及权限控制等核心业务逻辑；前端使用Vue框架开发用户界面，实现商品浏览、购物车操作、订单提交、二手交易、社区交流及个人中心等功能模块。系统通过角色划分支持注册用户、商家用户和管理员三类身份，各自具备相应的操作权限与功能视图，保障数据安全与业务流程有序进行。开发过程中遵循软件工程规范，完成需求分析、架构设计、编码实现及功能测试等环节，验证了平台在实际运行中的稳定性与可用性。结果表明，所采用的技术方案能够有效支撑摩托车垂直电商场景下的多样化需求，具备良好的可维护性与扩展潜力，为同类应用提供了可行的技术参考。
关键词：Spring Boot；Vue；前后端分离；摩托车商城；
Abstract This paper designs and implements a front-end and back-end separated motorcycle sales mall platform based on Java, Spring Boot, and Vue. The backend employs Spring Boot to construct RESTful services, handling core business logic such as user management, product information maintenance, order processing, and permission control. The frontend utilizes the Vue framework to develop the user interface, enabling features like product browsing, shopping cart operations, order submission, second-hand trading, community interaction, and personal center modules. The system supports three user roles—registered users, merchant users, and administrators—each with corresponding operational permissions and functional views, ensuring data security and orderly business processes. During development, software engineering standards were followed, completing requirements analysis, architectural design, coding implementation, and functional testing, validating the platform's stability and usability in practical operation. The results demonstrate that the adopted technical solution effectively supports diverse requirements in motorcycle vertical e-commerce scenarios, offering good maintainability and scalability potential, providing a feasible technical reference for similar applications. Keywords: Spring Boot; Vue; Front-end and Back-end Separation; Motorcycle Mall;
目 录 摘 要 Abstract 1 绪论 1.1 研究背景 1.2 研究意义 1.3 国内外研究现状 2 相关技术介绍 2.1 Java语言 2.2 SpringBoot框架 2.3 MySQL数据库 2.4 Vue框架 2.5 B/S模式 3 系统分析 3.1 系统可行性 3.1.1 技术可行性 3.1.2 经济可行性 3.1.3 操作可行性 3.2 系统功能需求 3.2.1 注册用户功能需求分析 3.2.2 前台商家用户功能需求分析 3.2.3 后台商家用户功能需求分析 3.2.4 管理员功能需求分析 3.3 非功能性需求 3.4 用户用例模型 3.4.1 注册用户用例图 3.4.2 （前台）商家用户用例图 3.4.3 （后台）商家用户用例图 3.4.4 管理员用例图 3.5 系统流程分析 3.5.1 注册登录流程图 3.5.2 数据添加流程图 3.5.3 数据修改流程图 3.5.4 数据删除流程图 3.5.5 数据搜索流程图 4 系统设计 4.1 系统架构设计 4.2 功能模块设计 4.3 数据库设计 4.3.1 概念结构设计 4.3.2 物理结构设计 5 系统实现 5.1 注册用户功能实现 5.1.1 用户注册模块 5.1.2 用户登录模块 5.1.3 首页模块 5.1.4 个人中心模块 5.1.5 线上商城模块 5.1.6 二手商品模块 5.2 （前台）商家用户功能实现 5.2.1 首页模块 5.3 （后台）商家用户功能实现 5.3.1 首页模块 5.3.2 商城管理模块 5.4 管理员功能实现 5.4.1 首页模块 5.4.2 二手商品管理模块 5.4.3 购买记录管理模块 5.4.4 退货记录管理模块 5.4.5 通知发布模块 5.5 测试目的方法 5.6 功能测试用例 5.7 测试结果分析 结 论 参考文献 致 谢 附 录 1绪论1.1研究背景随着互联网技术的持续演进，电子商务已深度融入人们日常生活，成为商品交易的重要渠道。传统线下摩托车销售模式受限于地域、信息传播效率及服务响应速度，在满足用户多样化需求方面面临挑战。与此同时，消费者对购车体验、信息透明度及售后服务的要求不断提高，推动行业向数字化、平台化方向转型。在此背景下，构建一个功能完善、操作便捷、安全可靠的线上摩托车销售平台，成为连接用户与商家、提升交易效率的关键路径。 近年来，前后端分离架构因其高内聚、低耦合的特性，被广泛应用于现代Web应用开发中。以Java语言为基础的Spring Boot框架凭借其快速开发、配置简化和生态成熟等优势，成为后端服务构建的主流选择；而Vue作为轻量级前端框架，以其组件化思想和高效渲染能力，显著提升了用户界面的交互体验。将二者结合，可为垂直领域电商平台提供稳定、灵活且易于维护的技术支撑，也为摩托车销售行业的数字化升级提供了可行的技术实现路径。 1.2研究意义本研究聚焦于摩托车垂直电商场景，通过构建集新车销售、二手交易、资讯发布、社区互动与订单管理于一体的综合性平台，有效整合分散的市场资源，打破信息壁垒，提升交易透明度与用户参与度。平台不仅为普通用户提供便捷的选购与交流空间，也为商家提供商品展示、订单处理与客户管理的统一入口，有助于优化运营流程、降低沟通成本，促进供需高效匹配。 从技术实践角度看，采用前后端分离架构进行开发，有助于明确职责边界，提升开发效率与系统可维护性。后端专注业务逻辑与数据安全，前端聚焦用户体验与界面表现，二者通过标准接口协同工作，既保障了系统的稳定性，也为后续功能扩展奠定良好基础。研究成果不仅验证了主流Web技术在特定行业应用中的适应性，也为其他垂直领域电商平台的建设提供了可复用的设计思路与实施经验，具有一定的推广价值与现实意义。 1.3国内外研究现状在电子商务领域，国内外对垂直类交易平台的研究与实践已取得显著进展。国外较早起步的电商平台如eBay、Amazon等，在摩托车及相关配件销售方面积累了丰富经验。eBay自20世纪90年代起便支持用户发布二手摩托车信息，其成熟的信用评价机制、支付保障体系和全球物流网络，为用户提供了安全便捷的交易环境。近年来，专门聚焦于摩托车领域的平台如RevZilla、Motorcycle.com等进一步细化服务，不仅提供新车与配件在线购买，还整合维修指南、车型评测、社区论坛等内容，形成“交易+内容+服务”的生态闭环。这些平台普遍采用模块化架构，后端以Java或.NET为基础支撑高并发访问，前端则通过现代JavaScript框架提升交互体验，体现出对用户需求深度挖掘与技术实现高度融合的特点。 国内电子商务发展虽起步稍晚，但增速迅猛，尤其在垂直电商方向展现出强大活力。以京东、淘宝为代表的综合平台早已开设摩托车及骑行装备专区，引入品牌官方旗舰店，保障正品与售后。与此同时，一批专注于机车文化的本土平台如“摩托邦”“哈罗摩托”等应运而生，除商品销售外，更强调用户社群运营与骑行生活方式的传播。例如，“哈罗摩托”通过整合车型数据库、用户口碑、线下活动报名等功能，构建起围绕摩托车用户的活跃社区，并逐步拓展至新车导购与二手车交易服务。在技术实现上，国内项目普遍采用Spring Boot构建后端服务，结合Vue或React开发前端界面，实现前后端解耦，提升开发效率与系统可维护性。尽管部分平台在订单履约、售后服务标准化等方面仍有提升空间，但整体已形成较为完整的线上摩托车消费生态，反映出市场对专业化、集成化数字平台的迫切需求与持续探索。 2 相关技术介绍2.1Java语言Java是一种广泛应用于企业级软件开发的面向对象编程语言，具有平台无关性、安全性高、稳定性强等优点[1]。通过JVM（Java虚拟机）机制，Java实现了“一次编写，到处运行”的特性，极大地提升了代码的可移植性和复用性，其丰富的类库支持和成熟的生态系统，使得Java在Web应用、分布式系统及大型后台服务开发中占据主导地位[2]。在面向摩托车销售商城系统中，Java作为主要开发语言，能够实现高效的服务器端逻辑处理和数据管理，为系统的功能实现提供了坚实的基础，同时Java的多线程和网络编程能力有助于系统的高并发处理，为多个用户提供服务。 2.2SpringBoot框架SpringBoot是一个开源框架，可以快速构建基于Spring的应用程序，通过简化配置和提供开箱即用的特性，极大地提升了开发效率[3]。SpringBoot框架安排了一系列默认的配置，支持开展自动化配置，减少了应用启动的复杂性，并集成嵌入式Web服务器，让开发者可以不依赖外部独立运行Java应用，无需借助外部容器[4]。SpringBoot框架的应用为面向摩托车销售商城系统的后端开发提供了强大的支持。开发者使用SpringBoot框架可以轻松创建RESTful风格的API，与前端进行数据交互，支持各种业务操作。数据访问层使用Spring Data JPA简单集成数据库，与MySQL进行操作，实现数据的持久化。安全管理方面利用Spring Security提供用户身份验证和授权功能，确保数据访问的安全性。 2.3MySQL数据库MySQL是一款流行的开源关系型数据库管理系统，因其性能稳定、易于使用、社区活跃而被广泛应用于各类Web应用系统中，支持标准SQL语句，即以结构化查询语言为基础，通过表格的形式存储数据，具备良好的事务处理能力和多用户并发访问能力[5]。同时MySQL占用资源较少，适合中小型项目的数据库需求。本系统采用MySQL作为核心数据存储工具，负责管理面向摩托车销售商城系统中的用户信息、业务数据和操作日志等关键信息，通过合理的数据库设计和优化，保障数据的安全性和一致性，并且MySQL支持对多用户的高并发操作，能够快速处理系统中的各种数据查询需求，为后端提供所需的数据服务[6]。 2.4Vue框架Vue 是一款轻量级、渐进式的前端JavaScript框架，专注于构建用户界面，具有学习成本低、开发效率高、组件化设计良好等特点，能够快速实现响应式数据绑定和动态页面交互[7]。Vue支持模块化开发，并与现代前端构建工具，如Webpack、Vite，兼容良好，适用于开发高性能、易维护的单页面应用（SPA）。在面向摩托车销售商城系统中，Vue框架被用于实现系统的前端界面，提供友好的交互体验，提升系统的交互体验与开发效率[8]。 2.5B/S模式B/S（Browser/Server）架构是一种基于浏览器的三层体系结构，用户通过浏览器即可访问服务器端的应用程序和数据资源，无需安装额外客户端，具有部署简单、维护方便、跨平台兼容性强等优势[9]。B/S架构降低了用户的客户端配置要求，用户只需要使用浏览器即可访问应用。该架构下，业务逻辑主要集中在服务器端处理，前端仅负责展示与交互，有利于提升系统的可扩展性和安全性[10]。本面向摩托车销售商城系统基于B/S架构进行设计与开发，使得系统能够在不同设备和操作系统上便捷访问，增强系统的可用性与灵活性。 3 系统分析3.1系统可行性3.1.1技术可行性系统采用Java语言结合Spring Boot框架构建后端服务，技术成熟、生态完善，具备良好的稳定性与可扩展性；前端使用Vue框架，支持组件化开发，提升界面交互体验与开发效率。前后端通过RESTful API进行通信，结构清晰、耦合度低，便于维护与迭代。相关开发工具、数据库（如MySQL）、中间件及部署环境均为主流且开源，技术门槛适中，开发资源丰富，能够有效支撑项目顺利实施。 3.1.2经济可行性项目所需软硬件资源均为通用配置，无需高昂的专用设备投入。开发过程主要依赖开源技术栈，大幅降低授权与许可成本。平台上线后可通过商品交易佣金、广告位出租、商家入驻服务等方式实现盈利，具备清晰的商业化路径。同时，线上运营模式减少了实体门店租金与人力开支，长期来看具有良好的成本控制优势和投资回报潜力。 3.1.3操作可行性系统功能设计贴合摩托车销售业务实际流程，界面简洁直观，用户学习成本低。注册用户、商家用户及管理员三类角色权限分明，操作逻辑符合各自使用习惯。后台管理功能集中、流程规范，便于商家高效处理订单与商品信息。整体交互流程顺畅，关键操作均有提示与反馈，保障用户在不同场景下均可顺利完成目标任务。 3.2系统功能需求根据用户在系统中的操作权限及功能需求的差异，本摩托车销售商城系统在设计过程中将用户角色划分为三大主要类别：注册用户、商家用户和管理员。其中，商家用户进一步细分为前台商家用户与后台商家用户——前者面向平台前端，主要进行商品展示、社区互动及基础信息维护；后者进入专属管理后台，负责商品发布、订单处理、售后管理等核心运营操作。注册用户可浏览商品、下单购买、参与社区交流并管理个人资料；管理员则拥有全局管控权限，负责用户管理、内容审核、公告发布及系统配置等整体运维工作。针对每种角色的具体功能职责，其详细的功能模块划分如下所述。 3.2.1注册用户功能需求分析我的购物车：允许用户收藏感兴趣的商品以便后续购买。 我的订单：查看历史订单详情及状态，支持订单跟踪。 我的地址：维护收货地址信息，简化下单流程。 个人首页：用户个性化的信息汇总页。 用户反馈：提交对平台的意见和建议，帮助改进服务。 订单配送：追踪订单物流状态，掌握配送进度。 交流社区：参与社区互动，分享经验。 收藏：保存感兴趣的帖子或商品链接。 评论管理：查看并回复自己发布的评论。 3.2.2前台商家用户功能需求分析个人首页：商家信息概览。 用户反馈：接收顾客评价与建议，持续优化服务质量。 订单配送：实时更新订单发货状态，提高客户满意度。 交流社区：作为行业专家分享专业知识，吸引目标客户。 收藏：保存感兴趣的行业资讯或竞争对手动态。 评论管理：管理顾客留言，维护品牌形象。 3.2.3后台商家用户功能需求分析线上商城：管理店铺内所有商品的信息更新、上下架操作。 分类列表：设置商品分类，方便顾客查找所需商品。 订单列表：查看所有订单详情，进行批量处理。 订单配送：更新订单的物流信息，确保顾客能及时收到商品。 订单售后：处理退货退款请求，保证售后服务质量。 3.2.4管理员功能需求分析3.3非功能性需求在面向摩托车销售商城系统的开发与设计过程中，除了满足基本的功能性需求外，还需要从多个维度考虑系统的非功能性要求，以保证系统的整体质量和用户体验。下面将从可靠性、可用性、性能、安全性等方面进行详细分析。 可靠性：系统应具备较高的稳定性和容错能力，在面对高并发访问或异常操作时能够保持正常运行。通过合理的异常处理机制和日志记录功能，确保系统在发生错误时可以快速定位问题并恢复服务，从而保障用户操作的连续性和数据的一致性。 可用性：系统界面应设计简洁、直观，操作流程清晰，便于不同层次的用户理解和使用。对于不同角色用户，应提供相应的引导提示和帮助信息以提升用户的操作效率和满意度。 性能需求：系统需支持多用户并发访问，并能在合理的时间内响应用户的请求。在典型负载条件下，页面加载时间应控制在合理范围内，关键业务操作的响应延迟不宜过高。 安全性：系统需具备完善的身份认证和权限控制机制，防止未经授权的访问和敏感信息泄露，并应对用户密码进行加密存储，对关键操作进行审计日志记录，确保系统具备一定的抗攻击能力和数据保护能力。 3.4用户用例模型3.4.1注册用户用例图注册用户作为核心参与者，可访问首页、交流社区、网站公告、摩托车资讯、聊天室、线上商城及二手商城等公共模块，实现信息浏览与互动；通过“商城管理”可操作购物车、订单和收货地址，完成购物流程；进入“个人中心”后，可进行个人主页维护、提交用户反馈、查看订单配送状态、参与社区讨论以及管理收藏与评论内容，全面支持用户的个性化使用需求。注册用户角色用例图如图3-1所示。
图3-1 注册用户用例图 3.4.2（前台）商家用户用例图前台商家用户在系统中的功能交互关系。作为平台内容提供者与服务参与者，前台商家用户可访问首页、交流社区、网站公告、摩托车资讯、聊天室、文心一言及二手商城等公共模块，实现信息获取与用户互动；同时具备发布和管理二手商品的能力，支持与注册用户进行沟通交流；通过“我的账户”和“我的主页”维护自身身份信息，并可在“个人中心”中查看个人首页、提交用户反馈、跟踪订单配送状态、参与社区讨论以及管理收藏与评论内容，全面支撑其在平台上的运营与社交需求。（前台）商家用户角色用例图如如图3-2所示。
图3-2 （前台）商家用户用例图 3.4.3（后台）商家用户用例图后台商家用户在系统中的功能交互关系。作为店铺运营主体，后台商家用户可访问后台首页以查看经营数据概览；通过交流管理模块对社区内容进行监控与维护；在商城管理模块中实现对线上商品的发布与维护、商品分类的查看、订单列表的查询与处理、订单配送状态的更新以及售后事务的管理，全面支持其在平台上的商品运营与订单服务流程。（后台）商家用户角色用例图如如图3-3所示。
图3-3（后台）商家用户用例图 3.4.4管理员用例图管理员在系统中的功能交互关系。作为平台的最高权限角色，管理员可访问后台首页以掌握系统整体运行状态；负责系统用户账号的管理与维护，对二手商品、购买记录、退货记录及用户反馈等关键业务数据进行统一监管；具备系统配置、网站公告发布、资源上传与管理、交流内容审核以及商城全局设置的权限，并可通过通知发布功能向用户推送重要信息，全面保障平台的安全性、规范性与高效运营。管理员角色用例图如如图3-4所示。
图3-4管理员用例图 3.5系统流程分析3.5.1注册登录流程图用户注册登录模块主要是为了方便用户和管理员能够安全地访问系统并管理自己的信息，用户根据提示输入注册信息进行注册操作以获得个人账号，在登录界面输入账号密码等信息进行系统登录，用户注册登录流程如下图所示。
图3-5 系统操作流程图 3.5.2数据添加流程图用户成功登录系统后，即可进行添加数据操作。添加的数据具有一个由系统自动生成的特定编号，用户可根据提示输入其余信息并提交，系统会对提交信息进行验证，验证通过则显示添加数据成功。数据添加流程如下图所示。
图3-6 数据添加流程图 3.5.3数据修改流程图用户成功登录系统后，可进行修改数据操作，流程与添加数据操作相似，数据修改流程如下图所示。
图3-7 数据修改流程图 3.5.4数据删除流程图当系统里面存在一些无效或过期的数据信息，系统支持相关的管理人员对这些数据进行删除操作，数据修改流程如下图所示。
图3-8 数据删除流程图 3.5.5数据搜索流程图用户可以通过输入关键字方式在系统大量的数据中检索所需的信息，在搜索框输入关键字确认查询后，系统会自动检索数据库并显示特定的数据信息，数据搜索流程如下图所示。
图3-9 数据搜索流程图 4 系统设计4.1系统架构设计本系统在架构设计上采用了经典的三层体系结构，分别为表现层、业务逻辑层以及数据访问层，并通过集群化管理方式实现高效的数据处理与并发访问。其中表现层主要承担用户交互功能，负责信息的展示与用户输入的接收，将用户操作传达给业务逻辑层处理，并最终将处理的结果反馈给用户，采用JavaScript等技术进行构建简洁和直观前端页面，以提升界面响应速度与用户体验。业务逻辑层作为系统的核心处理单元，主要负责各类核心业务规则的封装与执行，按需与数据访问层交互，以便获取或更新所需的数据，该层基于SpringBoot框架进行开发，借助其良好的模块化设计和自动化配置能力来保障系统业务流程的高效运行与稳定性。数据访问层则专注于与数据库之间的交互操作，包括数据的存储、查询与更新等功能，系统中采用JPA或MyBatis等持久层框架来实现对MySQL数据库的访问，确保数据操作的准确性、一致性与安全性。系统架构图如图4-1所示，展示了各层次之间的功能划分与数据流向关系。
图4-1 系统架构图 4.2功能模块设计系统以用户角色为基础，划分为注册用户、前台商家用户、后台商家用户和管理员四大模块，各角色拥有独立的功能分支。注册用户可访问首页、交流社区、资讯浏览、线上商城及个人中心等基础功能，支持商品购买与互动；前台商家用户在具备普通用户功能的同时，可发布二手商品并参与社区运营；后台商家用户专注于店铺管理，涵盖商品上架、订单处理与售后维护等核心业务；管理员则负责全局管控，包括用户管理、交易记录监管、公告发布、资源维护及通知推送等，实现平台的高效协同与统一运维。不同角色对应的具体功能模块如图4-2所示，面向摩托车销售商城系统功能设计能够确保各角色能够负责其特定职责。
图4-2 系统功能结构图 4.3数据库设计4.3.1概念结构设计通过提供清晰的系统总E-R图，可以使其他用户快速理解和分析复杂的系统结构，更加轻松地掌握了解系统的整体架构和各功能组件之间的联系。根据对面向摩托车销售商城系统中各类实体及其属性的分析，本面向摩托车销售商城系统总体E-R图如图4-3所示，以直观地展示各实体之间的关系。
图4-3 系统总体ER图 4.3.2物理结构设计依据前一节对面向摩托车销售商城系统的整体E-R关系图的分析，为了满足系统功能需求，需要创建多个数据表。下面将着重介绍几个核心数据库表的设计结构，详细阐述这些关键数据库表的设计细节，包括但不限于字段定义、数据类型及其相互间的关系，从而为系统的稳定运行提供坚实的基础。 表 4-1-access_token（登陆访问时长）
表 4-2-address（收货地址）
表 4-3-article（文章）
表 4-4-article_type（文章分类）
表 4-5-auth（用户权限管理）
表 4-6-business_user（商家用户）
表 4-7-cart（购物车）
表 4-8-code_token（验证码）
表 4-9-collect（收藏）
表 4-10-comment（评论）
表 4-11-follow（用户关注）
表 4-12-forum（论坛）
表 4-13-forum_type（论坛分类）
表 4-14-goods（商品信息）
表 4-15-goods_type（商品类型）
1.4 方差分析的基本假定对每一个总体（水平），要求： 二、单因素方差分析2.1 问题与模型设因素 AAA 有 kkk 个水平 A1,A2,…,AkA_1, A_2, \\ldots, A_kA1,A2,…,Ak，在水平 AiA_iAi 下进行 nin_ini 次独立试验，得到观测值 XijX_{ij}Xij（j=1,2,…,nij = 1, 2, \\ldots, n_ij=1,2,…,ni）。 线性统计模型： Xij=μi+εij,i=1,2,…,k;j=1,2,…,niX_{ij} = \\mu_i + \\varepsilon_{ij}, \\quad i = 1, 2, \\ldots, k; \\quad j = 1, 2, \\ldots, n_iXij=μi+εij,i=1,2,…,k;j=1,2,…,ni 其中 εij∼N(0,σ2)\\varepsilon_{ij} \\sim N(0, \\sigma^2)εij∼N(0,σ2) 且相互独立。 引入效应参数： 令总均值 μ=1n∑i=1kniμi\\mu = \\frac{1}{n}\\sum_{i=1}^{k} n_i \\mu_iμ=n1∑i=1kniμi（其中 n=∑i=1knin = \\sum_{i=1}^{k} n_in=∑i=1kni 为总观测次数），定义水平 AiA_iAi 的效应为： αi=μi−μ,i=1,2,…,k\\alpha_i = \\mu_i – \\mu, \\quad i = 1, 2, \\ldots, kαi=μi−μ,i=1,2,…,k 则模型改写为： Xij=μ+αi+εij,∑i=1kniαi=0\\boxed{X_{ij} = \\mu + \\alpha_i + \\varepsilon_{ij}, \\quad \\sum_{i=1}^{k} n_i \\alpha_i = 0}Xij=μ+αi+εij,i=1∑kniαi=0 待检验的假设为： H0:μ1=μ2=⋯=μk⇔H0:α1=α2=⋯=αk=0H_0: \\mu_1 = \\mu_2 = \\cdots = \\mu_k \\quad \\Leftrightarrow \\quad H_0: \\alpha_1 = \\alpha_2 = \\cdots = \\alpha_k = 0H0:μ1=μ2=⋯=μk⇔H0:α1=α2=⋯=αk=0 H1:μ1,μ2,…,μk 不全相等H_1: \\mu_1, \\mu_2, \\ldots, \\mu_k \\text{ 不全相等}H1:μ1,μ2,…,μk 不全相等 2.2 平方和分解（核心推导）2.2.1 定义基本统计量
重要恒等式验证： ∑i=1kniXˉi⋅=∑i=1kni⋅1ni∑j=1niXij=∑i=1k∑j=1niXij=nXˉ⋅⋅\\sum_{i=1}^{k} n_i \\bar{X}_{i\\cdot} = \\sum_{i=1}^{k} n_i \\cdot \\frac{1}{n_i}\\sum_{j=1}^{n_i} X_{ij} = \\sum_{i=1}^{k}\\sum_{j=1}^{n_i} X_{ij} = n\\bar{X}_{\\cdot\\cdot}i=1∑kniXˉi⋅=i=1∑kni⋅ni1j=1∑niXij=i=1∑kj=1∑niXij=nXˉ⋅⋅ 2.2.2 总离差平方和的分解定义总离差平方和： ST=∑i=1k∑j=1ni(Xij−Xˉ⋅⋅)2S_T = \\sum_{i=1}^{k}\\sum_{j=1}^{n_i}(X_{ij} – \\bar{X}_{\\cdot\\cdot})^2ST=i=1∑kj=1∑ni(Xij−Xˉ⋅⋅)2 分解过程： 对 Xij−Xˉ⋅⋅X_{ij} – \\bar{X}_{\\cdot\\cdot}Xij−Xˉ⋅⋅ 进行"加一项减一项"： Xij−Xˉ⋅⋅=(Xij−Xˉi⋅)+(Xˉi⋅−Xˉ⋅⋅)X_{ij} – \\bar{X}_{\\cdot\\cdot} = (X_{ij} – \\bar{X}_{i\\cdot}) + (\\bar{X}_{i\\cdot} – \\bar{X}_{\\cdot\\cdot})Xij−Xˉ⋅⋅=(Xij−Xˉi⋅)+(Xˉi⋅−Xˉ⋅⋅) 两边平方后求和： ST=∑i=1k∑j=1ni[(Xij−Xˉi⋅)+(Xˉi⋅−Xˉ⋅⋅)]2S_T = \\sum_{i=1}^{k}\\sum_{j=1}^{n_i}\\left[(X_{ij} – \\bar{X}_{i\\cdot}) + (\\bar{X}_{i\\cdot} – \\bar{X}_{\\cdot\\cdot})\\right]^2ST=i=1∑kj=1∑ni[(Xij−Xˉi⋅)+(Xˉi⋅−Xˉ⋅⋅)]2 展开平方项： ST=∑i=1k∑j=1ni(Xij−Xˉi⋅)2⏟SE+∑i=1k∑j=1ni(Xˉi⋅−Xˉ⋅⋅)2⏟SA+2∑i=1k∑j=1ni(Xij−Xˉi⋅)(Xˉi⋅−Xˉ⋅⋅)S_T = \\underbrace{\\sum_{i=1}^{k}\\sum_{j=1}^{n_i}(X_{ij} – \\bar{X}_{i\\cdot})^2}_{S_E} + \\underbrace{\\sum_{i=1}^{k}\\sum_{j=1}^{n_i}(\\bar{X}_{i\\cdot} – \\bar{X}_{\\cdot\\cdot})^2}_{S_A} + 2\\sum_{i=1}^{k}\\sum_{j=1}^{n_i}(X_{ij} – \\bar{X}_{i\\cdot})(\\bar{X}_{i\\cdot} – \\bar{X}_{\\cdot\\cdot})ST=SEi=1∑kj=1∑ni(Xij−Xˉi⋅)2+SAi=1∑kj=1∑ni(Xˉi⋅−Xˉ⋅⋅)2+2i=1∑kj=1∑ni(Xij−Xˉi⋅)(Xˉi⋅−Xˉ⋅⋅) 证明交叉项为零： ∑i=1k∑j=1ni(Xij−Xˉi⋅)(Xˉi⋅−Xˉ⋅⋅)=∑i=1k(Xˉi⋅−Xˉ⋅⋅)∑j=1ni(Xij−Xˉi⋅)\\sum_{i=1}^{k}\\sum_{j=1}^{n_i}(X_{ij} – \\bar{X}_{i\\cdot})(\\bar{X}_{i\\cdot} – \\bar{X}_{\\cdot\\cdot}) = \\sum_{i=1}^{k}(\\bar{X}_{i\\cdot} – \\bar{X}_{\\cdot\\cdot})\\sum_{j=1}^{n_i}(X_{ij} – \\bar{X}_{i\\cdot})i=1∑kj=1∑ni(Xij−Xˉi⋅)(Xˉi⋅−Xˉ⋅⋅)=i=1∑k(Xˉi⋅−Xˉ⋅⋅)j=1∑ni(Xij−Xˉi⋅) 由于 ∑j=1ni(Xij−Xˉi⋅)=∑j=1niXij−niXˉi⋅=niXˉi⋅−niXˉi⋅=0\\sum_{j=1}^{n_i}(X_{ij} – \\bar{X}_{i\\cdot}) = \\sum_{j=1}^{n_i}X_{ij} – n_i\\bar{X}_{i\\cdot} = n_i\\bar{X}_{i\\cdot} – n_i\\bar{X}_{i\\cdot} = 0∑j=1ni(Xij−Xˉi⋅)=∑j=1niXij−niXˉi⋅=niXˉi⋅−niXˉi⋅=0 因此交叉项为零，得到： ST=SE+SA\\boxed{S_T = S_E + S_A}ST=SE+SA 其中： SA=∑i=1k∑j=1ni(Xˉi⋅−Xˉ⋅⋅)2=∑i=1kni(Xˉi⋅−Xˉ⋅⋅)2S_A = \\sum_{i=1}^{k}\\sum_{j=1}^{n_i}(\\bar{X}_{i\\cdot} – \\bar{X}_{\\cdot\\cdot})^2 = \\sum_{i=1}^{k}n_i(\\bar{X}_{i\\cdot} – \\bar{X}_{\\cdot\\cdot})^2SA=i=1∑kj=1∑ni(Xˉi⋅−Xˉ⋅⋅)2=i=1∑kni(Xˉi⋅−Xˉ⋅⋅)2 SE=∑i=1k∑j=1ni(Xij−Xˉi⋅)2S_E = \\sum_{i=1}^{k}\\sum_{j=1}^{n_i}(X_{ij} – \\bar{X}_{i\\cdot})^2SE=i=1∑kj=1∑ni(Xij−Xˉi⋅)2 各项含义：
2.2.3 自由度的分解ST 的自由度：fT=n−1S_T \\text{ 的自由度：} \\quad f_T = n – 1ST 的自由度：fT=n−1 SA 的自由度：fA=k−1S_A \\text{ 的自由度：} \\quad f_A = k – 1SA 的自由度：fA=k−1 SE 的自由度：fE=n−kS_E \\text{ 的自由度：} \\quad f_E = n – kSE 的自由度：fE=n−k 验证： fA+fE=(k−1)+(n−k)=n−1=fTf_A + f_E = (k-1) + (n-k) = n – 1 = f_TfA+fE=(k−1)+(n−k)=n−1=fT ✓ 2.3 各平方和期望值的推导2.3.1 E(SE)E(S_E)E(SE) 的推导SE=∑i=1k∑j=1ni(Xij−Xˉi⋅)2S_E = \\sum_{i=1}^{k}\\sum_{j=1}^{n_i}(X_{ij} – \\bar{X}_{i\\cdot})^2SE=i=1∑kj=1∑<span class 行列式杂题第二弹文章目录
前言中档题 题目梗概行列式1.1 行列式多项式x³项系数求解 设四阶行列式 f ( x ) = ∣ x x 1 2 x 1 x 2 − 1 2 1 x 1 2 − 1 1 x ∣ f(x) = \\begin{vmatrix} x & x & 1 & 2x \\\\ 1 & x & 2 & -1 \\\\ 2 & 1 & x & 1 \\\\ 2 & -1 & 1 & x \\end{vmatrix} f(x)= x122xx1−112x12x−11x 求 f ( x ) f(x) f(x)中 x 3 x^3 x3项的系数。 行列式1.2 四阶三对角行列式行变换化上三角 计算四阶行列式 D 4 = ∣ 1 a 0 0 − 1 2 − a b 0 0 − 2 3 − b c 0 0 − 3 4 − c ∣ D_4 = \\begin{vmatrix} 1 & a & 0 & 0 \\\\ -1 & 2-a & b & 0 \\\\ 0 & -2 & 3-b & c \\\\ 0 & 0 & -3 & 4-c \\end{vmatrix} D4= 1−100a2−a−200b3−b−300c4−c 行列式1.3 矩阵线性变换求行列式 已知 α 1 , α 2 , α 3 \\boldsymbol{\\alpha}_1,\\boldsymbol{\\alpha}_2,\\boldsymbol{\\alpha}_3 α1,α2,α3是三维线性无关列向量， A \\boldsymbol{A} A是3阶矩阵，满足 { A α 1 = α 2 − 2 α 3 A α 2 = α 1 − α 2 + 2 α 3 A α 3 = 2 α 1 + α 2 \\begin{cases} \\boldsymbol{A}\\boldsymbol{\\alpha}_1 = \\boldsymbol{\\alpha}_2 – 2\\boldsymbol{\\alpha}_3 \\\\ \\boldsymbol{A}\\boldsymbol{\\alpha}_2 = \\boldsymbol{\\alpha}_1 – \\boldsymbol{\\alpha}_2 + 2\\boldsymbol{\\alpha}_3 \\\\ \\boldsymbol{A}\\boldsymbol{\\alpha}_3 = 2\\boldsymbol{\\alpha}_1 + \\boldsymbol{\\alpha}_2 \\end{cases} ⎩ ⎨ ⎧Aα1=α2−2α3Aα2=α1−α2+2α3Aα3=2α1+α2 求 ∣ A ∣ |\\boldsymbol{A}| ∣A∣。 1000题行列式2.1 三阶行列式函数零点个数判定 设 a i , b i , c i a_i,b_i,c_i ai,bi,ci为常数， f ( x ) f(x) f(x)不恒为0，讨论函数 f ( x ) = ∣ a 1 + x b 1 + x c 1 + x a 2 + x b 2 + x c 2 + x a 3 + x b 3 + x c 3 + x ∣ f(x) = \\begin{vmatrix} a_1+x & b_1+x & c_1+x \\\\ a_2+x & b_2+x & c_2+x \\\\ a_3+x & b_3+x & c_3+x \\end{vmatrix} f(x)= a1+xa2+xa3+xb1+xb2+xb3+xc1+xc2+xc3+x 的零点个数。 行列式2.2 n阶三对角行列式递推计算 计算n阶三对角行列式 D n = ∣ 2 1 0 ⋯ 0 1 2 1 ⋯ 0 0 1 2 ⋯ 0 ⋮ ⋮ ⋮ ⋱ ⋮ 0 0 0 ⋯ 2 ∣ D_n = \\begin{vmatrix} 2 & 1 & 0 & \\cdots & 0 \\\\ 1 & 2 & 1 & \\cdots & 0 \\\\ 0 & 1 & 2 & \\cdots & 0 \\\\ \\vdots & \\vdots & \\vdots & \\ddots & \\vdots \\\\ 0 & 0 & 0 & \\cdots & 2 \\end{vmatrix} Dn= 210⋮0121⋮0012⋮0⋯⋯⋯⋱⋯000⋮2 行列式2.3 两个同结构行列式的差 设 D 1 = ∣ 2 1 0 − 1 − 1 2 − 5 3 3 0 a b 1 − 3 5 0 ∣ , D 2 = ∣ 2 1 0 − 1 − 1 2 − 5 3 3 0 a b 1 − 1 1 0 ∣ D_1 = \\begin{vmatrix} 2 & 1 & 0 & -1 \\\\ -1 & 2 & -5 & 3 \\\\ 3 & 0 & a & b \\\\ 1 & -3 & 5 & 0 \\end{vmatrix}, \\quad D_2 = \\begin{vmatrix} 2 & 1 & 0 & -1 \\\\ -1 & 2 & -5 & 3 \\\\ 3 & 0 & a & b \\\\ 1 & -1 & 1 & 0 \\end{vmatrix} D1= 2−131120−30−5a5−13b0 ,D2= 2−131120−10−5a1−13b0 求 D 1 − D 2 D_1 – D_2 D1−D2。 行列式2.4 向量组线性变换求矩阵行列式（双解法） 设 A \\boldsymbol{A} A是3阶矩阵， α 1 , α 2 , α 3 \\boldsymbol{\\alpha}_1,\\boldsymbol{\\alpha}_2,\\boldsymbol{\\alpha}_3 α1,α2,α3是三维线性无关列向量，且满足 { A α 1 = α 1 + 2 α 2 + α 3 A α 2 = 2 α 1 + α 2 + α 3 A α 3 = α 1 + α 2 + 2 α 3 \\begin{cases} \\boldsymbol{A}\\boldsymbol{\\alpha}_1 = \\boldsymbol{\\alpha}_1 + 2\\boldsymbol{\\alpha}_2 + \\boldsymbol{\\alpha}_3 \\\\ \\boldsymbol{A}\\boldsymbol{\\alpha}_2 = 2\\boldsymbol{\\alpha}_1 + \\boldsymbol{\\alpha}_2 + \\boldsymbol{\\alpha}_3 \\\\ \\boldsymbol{A}\\boldsymbol{\\alpha}_3 = \\boldsymbol{\\alpha}_1 + \\boldsymbol{\\alpha}_2 + 2\\boldsymbol{\\alpha}_3 \\end{cases} ⎩ ⎨ ⎧Aα1=α1+2α2+α3Aα2=2α1+α2+α3Aα3=α1+α2+2α3 求 ∣ A ∣ |\\boldsymbol{A}| ∣A∣。 参考解析1.1考察点：行列式按行展开、多项式次数与系数分析 题目：设四阶行列式 f ( x ) = ∣ x x 1 2 x 1 x 2 − 1 2 1 x 1 2 − 1 1 x ∣ f(x) = \\begin{vmatrix} x & x & 1 & 2x \\\\ 1 & x & 2 & -1 \\\\ 2 & 1 & x & 1 \\\\ 2 & -1 & 1 & x \\end{vmatrix} f(x)= x122xx1−112x12x−11x 求 f ( x ) f(x) f(x)中 x 3 x^3 x3项的系数。 解： 将 f ( x ) f(x) f(x)按第1行展开，得四项： f ( x ) = x ⋅ A 11 + x ⋅ A 12 + 1 ⋅ A 13 + 2 x ⋅ A 14 f(x) = x\\cdot A_{11} + x\\cdot A_{12} + 1\\cdot A_{13} + 2x\\cdot A_{14} f(x)=x⋅A11+x⋅A12+1⋅A13+2x⋅A14 其中 A i j = ( − 1 ) i + j M i j A_{ij}=(-1)^{i+j}M_{ij} Aij=(−1)i+jMij为代数余子式，逐项分析 x 3 x^3 x3项的贡献： 第1项 x ⋅ A 11 x\\cdot A_{11} x⋅A11 M 11 M_{11} M11为去掉第1行第1列的三阶行列式： M 11 = ∣ x 2 − 1 1 x 1 − 1 1 x ∣ = x 3 − 4 x − 3 M_{11} = \\begin{vmatrix}x & 2 & -1 \\\\ 1 & x & 1 \\\\ -1 & 1 & x\\end{vmatrix} = x^3 – 4x – 3 M11= x1−12x1−11x =x3−4x−3 最高次为 x 3 x^3 x3，乘以 x x x后最高次为 x 4 x^4 x4，无 x 3 x^3 x3项贡献。 第2项 x ⋅ A 12 x\\cdot A_{12} x⋅A12 M 12 M_{12} M12为去掉第1行第2列的三阶行列式： M 12 = ∣ 1 2 − 1 2 x 1 2 1 x ∣ = x 2 − 2 x + 1 M_{12} = \\begin{vmatrix}1 & 2 & -1 \\\\ 2 & x & 1 \\\\ 2 & 1 & x\\end{vmatrix} = x^2 – 2x + 1 M12= 1222x1−11x =x2−2x+1 A 12 = − M 12 A_{12} = -M_{12} A12=−M12，乘以 x x x后得 − x 3 + 2 x 2 − x -x^3 + 2x^2 – x −x3+2x2−x， x 3 x^3 x3项系数为 − 1 -1 −1。 第3项 1 ⋅ A 13 1\\cdot A_{13} 1⋅A13 M 13 M_{13} M13为去掉第1行第3列的三阶行列式，最高次为 x 2 x^2 x2，乘以常数1后无 x 3 x^3 x3项贡献。 第4项 2 x ⋅ A 14 2x\\cdot A_{14} 2x⋅A14 M 14 M_{14} M14为去掉第1行第4列的三阶行列式： M 14 = ∣ 1 x 2 2 1 x 2 − 1 1 ∣ = 2 x 2 − x − 7 M_{14} = \\begin{vmatrix}1 & x & 2 \\\\ 2 & 1 & x \\\\ 2 & -1 & 1\\end{vmatrix} = 2x^2 – x – 7 M14= 122x1−12x1 =2x2−x−7 A 14 = − M 14 A_{14} = -M_{14} A14=−M14，乘以 2 x 2x 2x后得 − 4 x 3 + 2 x 2 + 14 x -4x^3 + 2x^2 + 14x −4x3+2x2+14x， x 3 x^3 x3项系数为 − 4 -4 −4。 合并所有 x 3 x^3 x3项系数： − 1 + ( − 4 ) = − 5 -1 + (-4) = -5 −1+(−4)=−5。
x 3 项的系数为 − 5 \\boxed{x^3项的系数为 -5} x3项的系数为−5 1.2考察点：行列式初等行变换、上三角行列式计算 题目：计算四阶行列式 D 4 = ∣ 1 a 0 0 − 1 2 − a b 0 0 − 2 3 − b c 0 0 − 3 4 − c ∣ D_4 = \\begin{vmatrix} 1 & a & 0 & 0 \\\\ -1 & 2-a & b & 0 \\\\ 0 & -2 & 3-b & c \\\\ 0 & 0 & -3 & 4-c \\end{vmatrix} D4= 1−100a2−a−200b3−b−300c4−c 解： 依次做行变换消去下三角元素：
化简后得到上三角行列式： D 4 = ∣ 1 a 0 0 0 2 b 0 0 0 3 c 0 0 0 4 ∣ D_4 = \\begin{vmatrix} 1 & a & 0 & 0 \\\\ 0 & 2 & b & 0 \\\\ 0 & 0 & 3 & c \\\\ 0 & 0 & 0 & 4 \\end{vmatrix} D4= 1000a2000b3000c4 主对角线元素相乘得结果： D 4 = 1 × 2 × 3 × 4 = 24 D_4 = 1 \\times 2 \\times 3 \\times 4 = 24 D4=1×2×3×4=24
D 4 = 24 \\boxed{D_4 = 24} D4=24 1.3考察点：相似矩阵行列式、向量组线性无关性、矩阵乘法 题目：已知 α 1 , α 2 , α 3 \\boldsymbol{\\alpha}_1,\\boldsymbol{\\alpha}_2,\\boldsymbol{\\alpha}_3 α1,α2,α3是三维线性无关列向量， A \\boldsymbol{A} A是3阶矩阵，满足 { A α 1 = α 2 − 2 α 3 A α 2 = α 1 − α 2 + 2 α 3 A α 3 = 2 α 1 + α 2 \\begin{cases} \\boldsymbol{A}\\boldsymbol{\\alpha}_1 = \\boldsymbol{\\alpha}_2 – 2\\boldsymbol{\\alpha}_3 \\\\ \\boldsymbol{A}\\boldsymbol{\\alpha}_2 = \\boldsymbol{\\alpha}_1 – \\boldsymbol{\\alpha}_2 + 2\\boldsymbol{\\alpha}_3 \\\\ \\boldsymbol{A}\\boldsymbol{\\alpha}_3 = 2\\boldsymbol{\\alpha}_1 + \\boldsymbol{\\alpha}_2 \\end{cases} ⎩ ⎨ ⎧Aα1=α2−2α3Aα2=α1−α2+2α3Aα3=2α1+α2<span class=\"vlist\" st vllm源码剖析13-vLLM 分布式推理-张量并行详解文章目录
张量并行（TP，Tensor Parallelism）的核心思路，是把模型中的大矩阵计算拆到多张 GPU 上完成。这样可以降低单张 GPU 的显存压力；当计算量足够大、通信开销可以被摊薄时，也可能提升 forward 吞吐。
后面我们就按 VocabParallelEmbedding、Column Parallel、Row Parallel 和 LM Head 这几类层，分别看它们在 forward 中什么时候触发 All-Reduce、All-Gather 或 gather。 下面将结合具体并行层，说明这些集合通信在 forward 中的触发位置，以及它们如何配合 TP 的切分方式完成计算。 一 AllReduce 原理All-Reduce 的目标，是将所有进程上的数据通过特定操作（如求和、取最大值）聚合后，把结果同步到每一个进程。常见底层实现包括 Ring AllReduce 和 Tree AllReduce。其最终目标，就是让每块 GPU 上的数据都变成汇总/归约后的同一个结果。 在 vLLM 的 TP 中，All-Reduce 典型出现在 Row Parallel 线性层和 Embedding 层：各 rank 先算出自己的部分结果，再通过 All-Reduce 求和，得到完整输出。具体我们会在本节的之后内容中讲到
1.1 Ring-AllReduce 算子原理Ring-AllReduce 的实现其实分为两个过程 Reduce-Scatter 和 All-Gather。 N N N 个进程参与通信，通常也对应 N N N 个 GPU。每个进程都有一份待归约的数据。Ring-AllReduce 会先把这份数据切成 N N N 个 chunk，并让这些进程在逻辑上组成一个环。 N − 1 N-1 N−1 轮后，每个进程都会得到一块已完成全局归约的结果分片。此时每个进程只保存完整结果的一部分。 N − 1 N-1 N−1 轮后，所有进程都能收集到全部归约分片，并把这些分片拼接成完整的 AllReduce 结果。 1.1.1 Reduce-Scatter以 4 个 GPU 设备为例，可将它们组织成一个逻辑环，使每个 GPU 只与相邻 GPU 交换数据。待归约的数据被切成 4 个 chunk，每轮通信时，各 GPU 同时发送一个 chunk、接收一个 chunk，并将收到的 chunk 与本地对应位置的 chunk 做逐元素相加。 第一次通信和累加完成后，某些位置的 chunk 已包含两个 GPU 的部分结果；这些更新后的 chunk 会在后续轮次中继续传递和累加。经过 N-1 轮（此处 N=4，即 3 轮）后，每个 GPU 会得到一个已完成全局累加的结果分片。 这里每次通信的数据量是一个 chunk。若原始数据量为 K，进程数为 N，则每个 chunk 的大小约为 K / N K/N K/N。Reduce-Scatter 阶段结束时，每个设备只持有完整结果的一部分；后续 All-Gather 阶段会将这些分片继续传播，使每个设备最终都获得完整结果。
1.1.2 All-GatherAll-Gather 通信操作依然遵循相邻 GPU 对应位置进行通讯的原则，但这一步不再做相加，而是将已归约好的分块拷贝到下一跳对应的位置上。All-Gather 以红色块作为起点，第一轮传递和填充完成后的示意图如下所示：
1.2 Ring AllReduce 的通信成本（重）假设有 N N N 个设备，原始数据总大小为 K K K，在一次 AllReduce 过程中，进行了 N − 1 N-1 N−1 次 Scatter-Reduce 操作和 N − 1 N-1 N−1 次 Allgather 操作，又因为每一次操作所需要传递的数据大小为 K / N K/N K/N，所以整个 AllReduce 过程所传输的数据大小为 2 ( N − 1 ) ∗ K / N 2(N-1) * K/N 2(N−1)∗K/N。随着 N 的增大，Ring AllReduce 通信算子的通信量可以近似为 2K。 在张量并行加速时，Ring AllReduce 的吞吐通常受到环上最慢链路和实现开销的共同影响。每次传输的数据块只有 K / N K/N K/N，所以当 N N N 增大时，单轮数据块变小。对于大张量，Ring AllReduce 容易形成流水线并行，通信效率通常较高；但对于小张量，固定调度开销和协议开销的占比会明显上升，实际有效带宽利用率会下降。 另外，Ring AllReduce 需要完成 2 × ( N − 1 ) 2 \\times (N-1) 2×(N−1) 次通信（两个阶段各 N − 1 N-1 N−1 次）。例如 8 GPU 需要 14 轮通信，即 Reduce-Scatter 和 All-Gather 各 7 轮。小张量本身的计算时间通常很短，但多轮通信带来的延迟累积会成为瓶颈。因此，Ring AllReduce 一般不适合小张量频繁同步、且对延迟要求很高的场景。 vLLM 在 CUDA all-reduce 路径中并不简单地固定使用一种传统 Ring-AllReduce。源码中会根据当前并行组、环境开关、world size、硬件拓扑、P2P 能力、张量大小、dtype、连续性等多种条件，尝试不同的通信实现，选择更合适的 all-reduce 路径，以降低延迟或提高带宽利用率。 以下伪代码用来抽象说明低延迟 AllReduce 的核心思想：各 GPU 通过预先建立的通信资源访问其他 rank 的数据，并完成逐元素求和，使每个 GPU 最终得到与 AllReduce(sum) 等价的全局规约结果。 def low_latency_all_reduce(x): # 同步和就绪检查省略。 for buf in peer_buffers: return out 从逻辑效果上看，每个 GPU 都会执行一次等价的规约过程，并得到与 AllReduce(sum) 相同的输出。和 Ring AllReduce 相比，这类低延迟路径不强调多轮分块传递，而是利用预先建立的通信资源、共享缓冲区等资源，尽量减少通信轮次和 kernel 调度次数，从而降低小张量频繁同步时的总延迟 二 Transformer 模型的张量并行decoder-only 架构的 LLM 中标准的 transformer 层如图 2 所示，其由一个自注意力（self-attention）模块和一个两层的多层感知机 (MLP)组成，可在这两个模块中分别引入模型并行（也叫张量并行）技术。 2.1 线性层权重不同切分方式张量并行的底层逻辑，可以先从矩阵乘法的拆分计算来理解。下面分别介绍线性层中的列并行和行并行，以及它们对应的矩阵形式。 其中 X X X 是输入， 是输入， 是输入，A 是权重矩阵。按这个数学形式， 是权重矩阵。按这个数学形式， 是权重矩阵。按这个数学形式，A$的第一维对应输入维，第二维对应输出维。后面说的“列并行”和“行并行”，都基于这个矩阵形式来理解 线性层是 Transformer 中最主要的 GEMM 来源之一，既出现在 MLP 中，也出现在 Attention 的 Q/K/V projection 和 output projection 中。开启张量并行后，vLLM 会让不同 TP rank 加载各自负责的权重分片，并在 forward 时分别执行本地 GEMM。 按不同方式切分权重后（TP=2）的线性层推理 forward 操作的可视化对比图如下图所示：
2.2 MLP 层的张量并行MLP 的张量并行实现相对 Self-Attention 简单。 2.2.1 拆分原理MLP 模块的第一个操作是通用矩阵乘法 (GEMM)，随后是一个 GeLU 非线性激活函数，计算公式如下所示。 Y = G e L U ( X A ) Y = GeLU(XA) Y=GeLU(XA) 后续最新的 Llama 及 qwen 系列的 llm 用的激活函数是 S i L U SiLU SiLU。 为了实现 GEMM 的并行计算。 第一个方法是将权重矩阵 A A A 按行拆分，同时将输入矩阵 X X X 按列拆分，如下图所示： X = [ X 1 , X 2 ] , A = [ A 1 A 2 ] X = [X_1, X_2],\\quad A = \\begin{bmatrix} A_1 \\\\ A_2 \\end{bmatrix} X=[X1,X2],A=[A1A2] 这种切分方式会得到 Y = GeLU ( X 1 A 1 + X 2 A 2 ) Y = \\text{GeLU}(X_1A_1 + X_2A_2) Y=GeLU(X1A1+X2A2)，因为 GeLU 是一个非线性函数，所以 GeLU ( X 1 A 1 + X 2 A 2 ) ≠ GeLU ( X 1 A 1 ) + GeLU ( X 2 A 2 ) \\text{GeLU}(X_1A_1 + X_2A_2) \\neq \\text{GeLU}(X_1A_1) + \\text{GeLU}(X_2A_2) GeLU(X1A1+X2A2)=GeLU(X1A1)+GeLU(X2A2)。因此这种方法在 GeLU 函数之前需要一个同步点。所谓同步点是指，在执行 GeLU 函数之前，需要等待各个设备的并行计算操作（即 X 1 A 1 X_1A_1 X1A1 和 X 2 A 2 X_2A_2 X2A2）都完成后，同时将各个设备输出的中间结果正确地聚合（synchronize）。可通过 reduce + broadcast 操作在 GPU0 上计算完整 X A XA XA，然后广播 GeLU ( X A ) \\text{GeLU}(XA) GeLU(XA) 到其他 GPU。 另一种方案是将权重矩阵 A A A 沿列方向切分为 A = [ A 1 , A 2 ] A = [A_1, A_2] A=[A1,A2]。这种切分方式使得每个 GPU 设备能独立完成部分 GEMM 运算，并应用 GeLU 激活函数。列切分权重方法的优势在于避免了前向传播中一次全局同步通信（行切分方案需先同步合并分块结果，再统一应用激活函数）。 [ Y 1 , Y 2 ] = [ GeLU ( X A 1 ) , GeLU ( X A 2 ) ] [Y_1, Y_2] = [\\text{GeLU}(X A_1), \\text{GeLU}(X A_2)] [Y1,Y2]=[GeLU(XA1),GeLU(XA2)]
2.2.2 拆分方式X A 1 XA_1 XA1 和 X A 2 XA_2 XA2）：每个 GPU 独立计算，无需通信。 Y 1 B 1 Y_1B_1 Y1B1 和 Y 2 B 2 Y_2B_2 Y2B2）之后，需要一次 All-Reduce 操作合并结果，再将结果输入 Dropout 层 – 之所以需要 All-Reduce 归约操作，是因为第二个线性层的权重是按行切分得到 Z 1 Z_1 Z1 和 Z 2 Z_2 Z2，所以需要执行加法操作得到最终的 Z。 – 具体来说，第二个 GEMM（如 Y 1 B 1 Y_1B_1 Y1B1 和 Y 2 B 2 Y_2B_2 Y2B2) 之后，需要一次 All-Reduce 操作合并结果（对 Y 1 B 1 Y_1B_1 Y1B1 和 Y 2 B 2 Y_2B_2 Y2B2求和）。 – 激活函数（如 GeLU）：本地计算，无需通信。 因为第二个线性层的权重是按行切分得到 Z 1 Z_ 1 Z1和 Z 2 Z_2 Z2，所以需要执行加法操作得到最终的 Z Z Z，即第二个 GEMM（如 $Y_1B_1$ 和 $Y_2B_2$之后，需要一次 All-Reduce 操作合并结果（对 $Y_1B_1$ 和 $Y_2B_2$求和）。以下的例子code/course12/simple_tp.py用于帮助大家理解这里的内容，其中X按列切分为 X 1 X_1 X1和 X 2 X_2 X2，A作为权重按行切分为 A 1 A_1 A1和 A 2 A_2 A2，两部分独立相乘之后得到 Y 1 Y_1 Y1和 Y 2 Y_2 Y2，随后再将它们相加就能得到最终的结果。 import numpy as np def gelu(x): # 1. 初始化参数 # 创建输入矩阵 X (B, H) 和 权重矩阵 A (H, D) # ========================================== # ========================================== # 假设我们有 2 个设备 (GPU)，将 A 按行切分，将 X 按列切分 # [切分权重矩阵 A] -> A1, A2 # [切分输入矩阵 X] -> X1, X2 # [并行计算] # [同步点 / All-Reduce] # [应用非线性激活函数] # ========================================== # 演示如果直接在局部做 GeLU 再相加是错误的 (数学原理验证)
MLP 的张量并行过程的形状变换公式拆解如下：
f f f运算符的实现示例： """ Implementation of f operator. g is similar to f with class g(torch.autograd.Function): 总结：当列并行与行并行级联使用时，在常见的 ColumnParallelLinear(gather_output=False) 接 RowParallelLinear(input_is_parallel=True) 的组合下，前级输出本来就是按最后一维切分后的局部结果，正好可以作为后级行并行所需的局部输入。 需要注意的是，先列后行的并行级联方式只是表示中间无需通信，并不意味着整个模块没有集合通信。RowParallelLinear 默认 reduce_results=True，通常会在自身输出侧执行 AllReduce，将各 GPU 的部分结果求和成完整输出。如果它的前面或后面还要与其他并行方式连接，也可能根据张量布局继续使用 AllGather、AllReduce、ReduceScatter 等集合通信操作。 先列后行的并行级联的 MLP 前向传播的可视化连接如下图所示: 2.2.3 MLA 层的通讯量分析(重)总结：MLP 层在 forward(前向推理时) 做一次 All-Reduce 操作，在 backward(前向推理时) 做一次 All-Reduce 操作。而 All-Reduce 的过程分为两个阶段，Reduce-Scatter 和 All-Gather，每个阶段的通讯量是相等的。假设输入张量大小为 [b, s, h]，数据类型为 fp16，则每次 All-Reduce 操作通讯量为 2bsh。 模型训练和推理阶段，MLP 层的张量并行通信量如下所示：
2.3 MHA 层的张量并行2.3.1 拆分原理多头注意力模块的结构如下图所示，可以看出，在设计上，MHA 层对于每个头（head），就有都有独立的 q/k/v 三个线性变换层以及对应的 self-attention 计算结构，然后将每个 head 输出的结果做拼接 concat，最后将拼接得到结果做线性变换得到最终的注意力层输出张量。
W Q W^Q WQ、 W K W^K WK、 W V W^V WV的维度都变成 (d_model, k_dim//2)。每个 head 上能独立做矩阵计算，最后将计算结果 concat起来即可。整个流程如下图所示：
可以发现，从多头注意力结构看，其计算机制真的是天然适合模型的张量并行计算，即每个头上都可以在每个设备上独立计算，即可以把每个头（也可以是 n 个头）的参数放到一块 GPU 上，最后将子结果 concat 后得到最终的张量。具体来说，多头注意力结构的张量并行计算过程拆解如下，实际模型中，会存在一个或多个 head 占用一块 GPU 的情况，且我们尽量保证 heads 总数能被 GPU 个数整除。 W Q W^Q WQ、 W K W^K WK、 W V W^V WV 做列并行切分 在多头注意力中，Q/K/V 投影可以按输出维，也就是 head 维进行切分。每个 TP rank 负责一部分 attention heads，并在本地完成这些 heads 对应的 Q/K/V 线性变换。因此，每个 rank 可以独立完成本地 heads 的 Q/K/V GEMM 和后续 attention 计算。在 Q/K/V 投影输出到本地 attention 计算之间，通常不需要立即进行跨 rank 集合通信。  vLLM 框架中 Attention 模块针对线性层的张量并行，实际代码实现也是先列后行线性层：
2.3.2 MHA 层的通讯量分析很明显上述的设计对 MLP 和自注意力层均采用了将两组 GEMM 运算融合的策略，从而消除了一个同步步骤，并获得了更好的扩展性。基于此技术方案，在一个标准 transformer 层中，前向传播只需执行两次 all-reduce 操作，反向传播也仅需两次 all-reduce（详见图 4）。
简单理解，这里的两次All-Reduce分别来自于self-attention层和MLP层 2.3.3 attention 中的张量并行实例在多头注意力中，Q、K、V 的输出维度可以看成由多个 attention heads 拼接而成。因此，当我们对 wq、wk、wv 的第二维进行切分时，本质上就是把不同的 heads 分配到不同的 tensor parallel rank 上计算。 在这个例子中，tp_size = 2，所以我们将 wq、wk、wv 按列切成两份。第一张卡（tp_rank = 0）负责前一半 heads；第二张卡（tp_rank = 1）负责后一半 heads。与 Q/K/V 投影不同，输出投影 wo 采用行切分。因为 attention 输出已经按 head 维度分布在不同的 rank 上，所以 wo 需要按输入维度（即第一维）进行切分： wq_sub1 = wq_weight[:, :hidden_dim // 2] 随后，将输入复制到每个 tensor parallel rank 上，并分别与本地切分后的 Q/K/V 权重进行矩阵乘法。下面是第一张卡上的计算结果，分别记作 q1、k1、v1： q1 = np.matmul(inputs, wq_sub1) # [bsz, seq_len, hidden_dim//2] 第二张卡的计算方式完全相同，只是使用的是 Q/K/V 权重的后一半切片。这样，每张卡都会得到自己负责的那部分 heads 对应的 q、k、v。接下来，每张卡使用本地的 q、k、v 独立计算 self-attention。由于标准多头注意力中不同 heads 之间是相互独立的，因此每个 rank 可以只计算自己负责的 heads，而不需要在 attention 计算阶段与其他 rank 通信。 得到局部 attention 输出之后，每张卡再将自己的 attention 输出与 wo 的对应行分块相乘。由于完整的输出投影可以拆成多个局部矩阵乘法结果之和，因此最后将各个 rank 的局部结果相加，就可以得到完整的 attention 输出： output_tp_parallel = attn_output1 @ wo_sub1 + attn_output2 @ wo_sub2 就是说，整个 self-attention 的张量并行过程可以概括为： Q/K/V 投影：列并行
Attention 计算：
输出投影 wo：行并行
import numpy as np if __name__ == '__main__': np.random.seed(42) # 设置随机种子，保证结果可复现 # 随机初始化权重矩阵 inputs = np.random.randn(bsz, seq_len, hidden_dim) print("===== 标准版本的多头注意力计算 =====") # 重塑为多头形式 # 调整维度顺序 # 注意力分数计算 # 应用softmax # 注意力输出计算 # 恢复原始维度 # 最终输出投影 print("===== 张量并行版本的多头注意力计算 =====") # GPU 1处理前half_heads个头 # 重塑为多头形式 # 调整维度顺序 # 计算注意力分数 # 应用softmax # 注意力输出计算 # 恢复原始维度 # GPU 2处理后half_heads个头 # GPU 2上的计算 # 重塑为多头形式 # 调整维度顺序 # 计算注意力分数 # 应用softmax # 注意力输出计算 # 恢复原始维度 # 合并结果 (相当于在head维度上concatenate) print(np.mean(np.abs(output – output_tp_parallel))) 2.4 Embedding 层的张量并行Transformer 语言模型的输出侧通常会通过 LM head 将 hidden states 投影到词表维度。这里需要区分两个概念：LM head 的权重矩阵通常是 [vocab_size, hidden_size]，而计算得到的 logits 通常是 [batch_size 或 token 数, vocab_size]。由于当前 Transformer 语言模型的词汇表通常至少有数万个 token，因此对 LM head 这类大词表投影进行张量并行，通常可以减少单张 GPU 上的计算量和显存占用。 在 vLLM 中，输入 embedding 和输出 LM head 都采用按词表维度切分的方式。需要注意的是，输入 embedding 的通信方式和输出 LM head 不一样： 2.4.1 输入嵌入层Embedding 层开启 TP 时，会将输入嵌入层的权重矩阵 E E E（尺寸为 [vocab_size, hidden_size]）按词汇维度拆分。由于词汇维度对应矩阵的第 0 维，所以可以理解为“按行拆分”：每个 GPU 只保存一段 token id 范围对应的 embedding 行。例如 TP=2 时，可以把词表分成两个连续范围，rank 0 负责前半段 token，rank 1 负责后半段 token。 因为每个分块只包含嵌入表的一部分，所以每张卡只能直接查到属于自己词表范围内的 token。对于不属于当前 rank 的 token，vLLM 会先通过 mask 将其排除，并把对应位置的 embedding 输出置为 0。随后，各个 rank 对局部 embedding 结果执行一次 all-reduce。由于同一个 token 只会在负责它的 rank 上产生非零 embedding，all-reduce 求和后就能得到完整的输入嵌入结果。 简单理解这个过程：假设现在有两个 GPU，每个 GPU 负责一半词表。输入请求的 seq=2，input_ids=[31, 33]。假设 token 31 属于 GPU 0 的词表范围，token 33 属于 GPU 1 的词表范围。那么在 GPU 0 上，token 31 能查到真实 embedding，token 33 不属于本卡词表范围，会被置为 0： GPU 0: 在 GPU 1 上则相反，token 31 会被置为 0，token 33 能查到真实 embedding： GPU 1: 最后对两个 GPU 的结果做 all-reduce 求和，就得到完整的输入嵌入： [ 也就是说，输入 embedding 的 TP 流程可以概括为：先判断每个 token 是否属于当前 rank 的词表范围；属于本 rank 的 token 正常查表；不属于本 rank 的 token 先 mask 掉并把输出置 0；最后通过 all-reduce 把各个 rank 的局部结果相加，恢复完整的输入 embedding。 import numpy as np # ========================================== # 完整的 Embedding 表 (模拟 Ground Truth) # 输入 Token IDs (Batch=2, Seq=2) print("输入 IDs:\\n", Input_IDs) # ========================================== # ========================================== # [切分权重] 按词汇维度切分 (Row Parallel in terms of Matrix, # GPU 1: 负责 ID 0-4 # GPU 2: 负责 ID 5-9 def parallel_embedding_forward(input_ids, local_weight, start_idx, end_idx): # 2. 将全局 ID 映射为本地 ID (Offset) # 3. 为了避免索引越界，将不属于自己的 ID 置为 0 # 4. 查表 (Lookup) mask_expanded = mask[:, :, np.newaxis] # 5. 只有属于自己的 ID 保留 Lookup 结果，其他的变成 0.0 return final_local_output # — GPU 1 计算 — # — GPU 2 计算 — # ========================================== print(f"\\n验证结果: {np.allclose(Output_Ref, Output_Fused)}") 2.4.2 输出嵌入层对于输出 LM head，先考虑一种朴素的完整 logits 汇总方案。假设 hidden states 为 X X X，输出权重矩阵按词表维度切分为 E 1 , E 2 E_1, E_2 E1,E2。若 E i E_i Ei 的形状沿用 embedding 权重的表示，即 [vocab_size/tp_size, hidden_size]，那么每个 rank 上的本地 logits 应写成： Y i = X E i T Y_i = X E_i^T Yi=XEiT 两个 rank 并行计算后，可以得到词表分片上的 logits： [ Y 1 , Y 2 ] = [ X E 1 T , X E 2 T ] [Y_1, Y_2] = [X E_1^T, X E_2^T] [Y1,Y2]=[XE1T,XE2T]，接下来再通过 gather 或 all-gather 将这些本地 logits 按词表维度拼接起来，得到完整 logits： Y = gather/all-gather ( [ Y 1 , Y 2 ] ) Y = \\text{gather/all-gather}([Y_1, Y_2]) Y=gather/all-gather([Y1,Y2]) 三 vLLM 中的张量并行3.1 vLLM 中张量并行如何使用使用 vLLM 启动模型服务时，如果模型太大，无法放入单个 GPU，但可以放入单个节点中的多个 GPU，就可以使用张量并行。只考虑 TP、没有叠加 PP、DP、DCP 等其他并行维度时，tensor_parallel_size 通常就等于你希望用于单个模型副本的 GPU 数量。例如，单节点有 4 个 GPU，可以将张量并行大小设置为 4。 对于多 GPU 的离线推理，可以在 LLM 类中设置 tensor_parallel_size 为所需的 GPU 数量。例如，要在 4 个 GPU 上运行推理：： from vllm import LLM 对于多 GPU 服务，也就是在线推理，可以在启动服务器时包含 –tensor-parallel-size。例如，在 4 个 GPU 上运行 API server： # 前提是安装成功了 vllm，在可通过下述命令启动多 GPU 服务 在 vLLM 的多进程执行路径中，前端进程通过 ZMQ socket 与后台 EngineCore 通信。开启 DP 时，vLLM 会按 DP 模式管理多个 EngineCore；MoE 的 DP 场景会使用 DPEngineCoreProc，非 MoE 的 DP rank 则更接近多个相互独立的 EngineCore。 如果同时开启 DP 和 TP，可以把每个 DP rank 理解为一个模型副本，副本内部再按 tensor_parallel_size 切分到多个 GPU。只开启 DP+TP 时，1 个 DP 副本内部的 GPU 数量等于 TP 数；如果还叠加 PP（pipeline parallel）或 prefill context parallel，则还要继续乘上这些并行维度。 3.2 vLLM 中的分布式资源管理vLLM 中与分布式资源管理相关的核心逻辑主要位于 vllm/distributed 目录。其中，distributed/parallel_state.py 负责分布式并行状态管理。它封装了 PyTorch ProcessGroup 等底层通信机制，为推理过程提供统一的并行组管理和通信接口。 工作流如下： 在 initialize_model_parallel() 中，vLLM 会根据当前 world_size、rank，以及 TP、PP、DP、prefill/decode context parallel、EP 等配置，构造全局 rank 布局，布局顺序是：ExternalDP x DP x PP x TP 随后，initialize_model_parallel() 通过 reshape、transpose 等操作生成不同并行维度的 group_ranks。例如，TP 组会把同一个 PP stage 内的 rank 放到一起；PP 组会把同一个 TP rank、跨不同 PP stage 的 rank 放到一起。 init_model_parallel_group() 更偏底层：它接收已经计算好的 group_ranks，并创建对应的 GroupCoordinator。GroupCoordinator 是对 PyTorch ProcessGroup 的封装，会为这些 rank 创建 device 通信组，并记录当前进程在组内的 rank_in_group、world_size 等信息。后续的 all_reduce、all_gather、send/recv 等操作，都会通过对应的 GroupCoordinator 在正确的 rank 范围内执行。 vLLM 使用模块级全局变量保存并行组状态。例如，_TP 记录张量并行组，_PP 记录流水线并行组，_DP 记录数据并行组；当前版本还包含 _DCP、_PCP、_EP、_EPLB 等组状态。其中，_EP 主要用于 MoE 场景，dense model 通常不会创建 EP group。后续代码可以通过 get_tp_group()、get_pp_group()、get_dp_group() 等函数获取对应通信组。 例如，假设有 8 张 GPU，并设置：
此时world_size = 8，avLLM 会先将 global rank 组织成一个二维结构： PP stage 0: GPU0 GPU1 然后生成 TP 和 PP 两类通信组： TP groups：
PP groups：
其中，TP 组负责同一个 PP stage 内的张量并行通信。例如，[GPU2, GPU3] 共同负责 stage 1 的模型层，RowParallelLinear 的 AllReduce 只会在 GPU2 和 GPU3 之间发生。 PP 组负责同一个 TP rank 跨不同 PP stage 的通信。例如，[GPU1, GPU3, GPU5, GPU7] 表示 TP rank 1 这一路的流水线通道。前向执行时，stage 0 的 GPU1 会将对应激活发送给 stage 1 的 GPU3，随后再沿 GPU3 -> GPU5 -> GPU7 向后传递 _TP = init_model_parallel_group(group_ranks, 四种并行组作用如下：
实例理解，TP + PP 并行组的概念。在 8 卡 GPU 上配置 流水线并行度（PP）为 4 和 张量并行度（TP）为 2 时，并行分组逻辑如下：
# 共 4 个张量并行（TP）组，每组在 2 张卡上切分模型参数
模拟分布式并行分分组算法可以通过下述示例代码（CPU模拟，不需要多机）来模拟分布式并行分分组算法，得到不同 gpu 设备和不同推理配置下的并行分组信息。rank网格简化为 [ExternalDP, DP, PP, TP]。 import torch def compute_groups(world_size, tp, pp, dp): # TP 组 # PP 组 # DP 组（模型内部 DP） # EP 组（同 PP stage 下合并 DP×TP） return tp_groups, pp_groups, dp_groups, ep_groups print("TP groups:", tp_groups) if __name__ == "__main__": 输出结果如下所示: TP groups: [[0, 1, 2, 3], [4, 5, 6, 7], [8, 9, 10, 11], [12, 13, 14, 15]] 这里把这些 rank 按照不同类型拆分到 4 维上。这里的 ext_dp_size 指的是本系统之外的一层数据并行维度，在本节课中不做深究。
3.3 vLLM 的并行线性层vllm/model_executor/models/qwen3.py 代码中的 Qwen3Attention 模块的核心网络层组成如下所示: Qwen3Attention (继承自 nn.Module) 其中 qkv 线性层使用 TP 是按列切分权重，o 线性层是按行切分权重。 ColumnParallelLinear 类源码剖析(列并行)ColumnParallelLinear 和 RowParallelLinear 类的基类是 LinearBase，它为所有并行及量化的线性层提供了基础框架和通用属性，其核心支持了普通与量化两种线性层实现。 ColumnParallelLinear 类是一种实现列并行（Column Parallelism）的线性层。其中权重 A 按第二维（输出维度）列切分，被划分为多个子矩阵： A = [ A 1 , A 2 , … , A p ] A = [A_1, A_2, \\dots, A_p] A=[A1,A2,…,Ap]，每个 GPU 只存储并计算其中一块 A i A_i Ai，对应输出 Y i = X A i Y_i = X A_i Yi=XAi。如果需要完整的输出，多个 GPU 上的 Y i Y_i Yi<span class= 鸿蒙分布式能力与超级终端实战：设备发现·数据同步·任务流转 |
第二次累加完成后的示意图如下，同样，被更新的数据块，会作为下一次传递和累加的起点，继续参与下一轮的通信和计算。
第三次累加完成后的示意图如下： 
经过 3 次环形传递和规约后，每块 GPU 上都有一块数据拥有了对应位置完整的累加聚合（下图中红色块）。此时，Reduce-Scatter 通信阶段结束，进入 All-Gather 通信阶段。目标是将红色块继续沿环传播，并填充到其余 GPU 对应的位置上，使所有 GPU 最终都拥有全部数据。
同样的经过 3 轮更新，使得每块 GPU 上都汇总到了完整的数据，变成如下形式: 
基于 transformer 网络 pytorch 代码的基础上，通常只需添加几个同步操作代码（synchronization primitives），就可实现一个简单的模型并行方案。下文我将会描述 Megatron-LM 的张量并行算法原理，以及在 transformer 模型中的应用。
将第一个线性层的权重按照列并行方式切分后，第二个线性层的权重矩阵 B 自然沿着行方向拆分，使其能够直接处理来自 GeLU 层的输出而无需任何通信，如图 3a 所示。
vLLM 框架中 MLP 模块针对线性层的张量并行，它的实际代码实现也是先列后行线性层： 
下图展示了当 num_attention_heads = 2 时 attention 层的 Q/K/V 线性变换的并行计算方法。对每一块权重，我们都沿着列方向（k_dim）维度切割一刀。此时每个 head 上的 
和 MLP 模块类似，模型训练和推理阶段，MHA 层的张量并行通信量如下所示：
结合图表与分组信息，可以这样理解 TP + PP 并行组内的交互：
