Spark SQL 重分区：并行度调优的 TaoToken 实践指南

1. Spark SQL 重分区到底在改什么：并行度、分区数与数据倾斜的真实关系

Spark SQL 里的重分区，说白了就是一句话：把数据重新分堆，顺便决定后面有多少个 Task 一起干活。repartition和coalesce这两个算子，一个能增能减、必然触发 Shuffle，一个只能减、尽量不 Shuffle。很多做 ETL 的朋友一看到任务慢，第一反应就是repartition(1000)拉满并行度，结果发现更慢了——因为 Shuffle 写盘的开销把并行带来的收益全吃掉了。

我先把核心概念摆清楚，不然后面调参全是玄学。

分区数（Partition Num）决定了一个 Stage 里能同时跑多少个 Task。分区数太少，集群 CPU 空转，一个 Task 扛几百万行；分区数太多，每个 Task 调度开销、序列化开销、小文件问题全来了。并行度（Parallelism）在 Spark SQL 语境下，通常就等于某个 Stage 的分区数，但要注意它会被spark.sql.shuffle.partitions这个默认值（默认 200）悄悄接管。

这里有个特别容易被忽略的点：Spark SQL 的并行度是分阶段看的。一个group by查询，物理计划里会出现两次Exchange——一次是partial_count的局部聚合，一次是hashpartitioning的全局聚合。你repartition加在哪一层，影响的是哪一段的 Task 数量，完全不一样。

举个我踩过的坑：一个日增 8000 万行的订单表做group by city，默认 200 分区，跑 40 分钟。我直接repartition(2000)，结果跑了 55 分钟。为什么？因为数据本身按 city 分布极不均匀，北上广深四个城市占了 70% 的数据量，repartition用的是 RoundRobin 轮询，它不解决倾斜，只是把行数平均分，但同一个 city 的 key 还是全落到同一个 reducer 上。真正该做的是加盐或者用REPARTITION(col)按列重分。

所以重分区的双向影响是这样的：

操作	分区数变化	是否 Shuffle	适用场景
coalesce(n)	只能减少	否（窄依赖）	减少输出文件、小结果集合并
repartition(n)	可增可减	是（宽依赖）	提升并行度、打散倾斜
repartition(n, col)	可增可减	是（按列哈希）	按业务键重分布
repartitionByRange(n, col)	可增可减	是（范围分区）	有序数据、范围查询

coalesce不减到 1 的时候不触发 Shuffle，这是它最大的优势，但代价是上游 Stage 的并行度不会变，只是下游读取时合并。很多人以为coalesce(10)能让整个作业只用 10 个 Task，其实上游该多少还是多少。

那分区数到底怎么算？我总结了一个经验公式，实测下来比较稳：

目标分区数 = max( 集群可用核数 × 2 ~ 3, 数据量(GB) × 1024 / 单分区目标大小(MB) )

单分区目标大小一般控制在128MB ~ 256MB，跟 HDFS Block 对齐。比如 500GB 数据，按 200MB 一个分区，就是500 × 1024 / 200 ≈ 2560个分区。但如果你集群只有 100 核，那 2560 个分区意味着每个核要跑 25 个 Task，调度开销会很明显，这时候要权衡。

还有一个隐藏参数：spark.sql.files.maxPartitionBytes，默认 128MB，它决定读文件时怎么切分。如果你读的是小文件，这个参数帮不上忙，得先用repartition合并。

理解了这些，你才能明白为什么"增大并行度"和"减少并行度"不是对立的，而是看你的瓶颈在哪。CPU 打满但 Task 数少 → 增大；Shuffle 写盘爆炸、Task 数远超核数 → 减少；数据倾斜 → 按列重分区或加盐。下面我带你把这套逻辑落到可复制的配置上。

2. TaoToken 统一通道前置准备：Base URL、API Key 与 Model ID 三件套

调优验证这件事，最怕的就是环境不统一——本地跑通了，集群上因为网关、鉴权、模型版本不一致，结果对不上。我现在的做法是：所有 Spark SQL 的调优实验，都通过 TaoToken 统一通道走，这样 Base URL、Key、Model ID 三件套固定，换机器换集群都不用改代码。

先说清楚 TaoToken 是什么：它是一个统一的模型调用通道，把不同模型的接入方式收敛成一套 OpenAI 兼容的接口。对做数据工程的人来说，它的价值在于——你可以在 Spark 作业里直接调用模型做数据质量校验、SQL 生成、异常检测，而不用为每个模型单独维护一套 SDK 和鉴权逻辑。

第一步，拿 API Key。打开https://taotoken.net/api-keys，登录后创建一个新 Key。建议按用途命名，比如spark-etl-tuning，方便后面排查是哪个作业在调用。Key 只在创建时显示一次，复制下来存到安全的地方。

第二步，确认 Base URL。TaoToken 的 API 入口是：

https://taotoken.net/api

注意这里不要加任何 UTM 参数，API 调用要的是干净的地址。如果你在浏览器里访问官网了解产品，用这个带来源的地址：

https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

第三步，选 Model ID。这是最容易出错的地方。TaoToken 支持多个模型，Model ID 必须和你在控制台看到的一致。比如做代码相关的调优辅助，可以用claude-code系列；做通用对话验证，用gpt-4o之类的。具体可用列表在https://taotoken.net/doc里有。

三件套凑齐后，先别急着写 Spark 代码，用 curl 验证一下通道是否通：

curl -X POST https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "gpt-4o", "messages": [{"role": "user", "content": "ping"}], "max_tokens": 10 }'

返回里有choices[0].message.content就说明通道正常。这一步很重要，因为后面 Spark 里报的错，很多其实是 Key 或 Model ID 写错了，而不是 Spark 本身的问题。

为什么要在 Spark 场景里用统一通道？因为调优验证往往需要对比多次运行的结果。比如你改了repartition参数，想确认并行度真的变了，可以让模型帮你解析 Spark UI 的 JSON 输出，或者对比两次explain()的物理计划差异。如果每次调用都要换 Key、换地址，实验根本没法复现。

还有一个实际好处：Coding Plan。如果你长期做 Spark 调优，需要反复让模型分析执行计划、生成测试 SQL，用 Coding Plan 比按次调用划算得多。入口在https://taotoken.net/coding-plan。

环境变量建议这样设，避免 Key 硬编码进代码：

export TAOTOKEN_API_KEY="sk-xxxxxxxx" export TAOTOKEN_BASE_URL="https://taotoken.net/api" export TAOTOKEN_MODEL="gpt-4o"

Windows 下用set或者写进系统环境变量。设完之后echo $TAOTOKEN_API_KEY确认一下。

这里插一句：不要把 Key 提交到 Git。我见过太多人把 Key 写进spark-defaults.conf然后推到公共仓库。正确做法是用--conf spark.executorEnv.TAOTOKEN_API_KEY=xxx在提交时注入，或者用密钥管理服务。

前置准备做完，下面进入真正的配置环节。我会给你可以直接复制的 JSON、TOML 和 Spark 配置片段。

3. 可复制的重分区配置：JSON/TOML/settings 片段与分区数计算

这一节是全文的核心，我尽量给能直接抄的东西。先说清楚：Spark SQL 的 repartition hint 从 2.4 开始支持，如果你还在 2.3 或更早，hint 会被静默忽略，不报错但也不生效——这是最坑的地方，很多人以为加了 hint 就完事，其实根本没起作用。

SQL Hint 写法：

-- 增大并行度到 1000 SELECT /*+ REPARTITION(1000) */ city, count(*) FROM orders GROUP BY city; -- 按列重分区，解决倾斜 SELECT /*+ REPARTITION(500, city) */ city, count(*) FROM orders GROUP BY city; -- 减少分区，合并小文件 SELECT /*+ COALESCE(50) */ * FROM orders WHERE dt = '2024-01-01'; -- 范围重分区，适合有序数据 SELECT /*+ REPARTITION_BY_RANGE(200, order_date) */ * FROM orders;

注意REPARTITION(500, city)和REPARTITION(500)的区别：前者按 city 哈希，同一个 city 的数据保证落到同一分区；后者是 RoundRobin，纯轮询。解决倾斜必须用带列的那个。

Spark 配置片段（spark-defaults.conf）：

# Shuffle 分区数，这是最关键的参数 spark.sql.shuffle.partitions=800 # 自适应执行，Spark 3.x 强烈建议开启 spark.sql.adaptive.enabled=true spark.sql.adaptive.coalescePartitions.enabled=true spark.sql.adaptive.skewJoin.enabled=true # 单分区目标大小，AQE 合并分区时参考 spark.sql.adaptive.advisoryPartitionSizeInBytes=256MB # 读文件时的最大分区字节数 spark.sql.files.maxPartitionBytes=128MB # 动态资源分配 spark.dynamicAllocation.enabled=true spark.dynamicAllocation.minExecutors=5 spark.dynamicAllocation.maxExecutors=100

如果用 Spark Submit 提交，命令行参数：

spark-submit \ --master yarn \ --deploy-mode cluster \ --conf spark.sql.shuffle.partitions=800 \ --conf spark.sql.adaptive.enabled=true \ --conf spark.sql.adaptive.coalescePartitions.enabled=true \ --conf spark.sql.adaptive.advisoryPartitionSizeInBytes=256MB \ --conf spark.executorEnv.TAOTOKEN_API_KEY=$TAOTOKEN_API_KEY \ --conf spark.executorEnv.TAOTOKEN_BASE_URL=https://taotoken.net/api \ --class com.example.SparkTuningJob \ spark-tuning.jar

分区数计算公式，我把它写成一个可执行的 Scala 函数：

def calcPartitionNum( dataSizeGB: Double, clusterCores: Int, targetPartitionMB: Int = 200 ): Int = { // 按数据量算 val bySize = (dataSizeGB * 1024 / targetPartitionMB).toInt // 按核数算，每个核 2~3 个 Task val byCores = clusterCores * 3 // 取较大值，但不超过 bySize 的 2 倍 math.min(math.max(bySize, byCores), bySize * 2) }

举个例子：500GB 数据，100 核集群，目标分区 200MB。

• bySize = 500 * 1024 / 200 = 2560
• byCores = 100 * 3 = 300
• 结果 =min(max(2560, 300), 5120) = 2560

所以设spark.sql.shuffle.partitions=2560。但如果你开了 AQE，其实可以设大一点，让 AQE 自动合并：

spark.sql.shuffle.partitions=4000 spark.sql.adaptive.coalescePartitions.enabled=true spark.sql.adaptive.advisoryPartitionSizeInBytes=256MB

AQE 会在运行时把小的分区合并到 256MB 左右，这样既避免了手动算不准，又能动态适应数据量波动。

TaoToken 的 TOML 配置（如果你用配置文件管理）：

[taotoken] base_url = "https://taotoken.net/api" api_key_env = "TAOTOKEN_API_KEY" model = "gpt-4o" timeout_seconds = 60 [spark.tuning] shuffle_partitions = 2560 adaptive_enabled = true advisory_partition_size = "256MB" skew_join_enabled = true

关于coalesce和repartition的选择，我给个决策表：

场景	用哪个	原因
输出文件太多，想合并	coalesce(n)	不 Shuffle，快
上游 Task 太少，CPU 空转	repartition(n)	必须 Shuffle 才能增并行
数据倾斜，某几个 key 特别大	repartition(n, key)	按 key 哈希重分布
需要全局排序	repartitionByRange(n, col)	范围分区保证有序
不确定数据量	开 AQE，设大 shuffle.partitions	让 Spark 自己合并

还有一个细节：repartition之后的分区数会一直传递到下一个 Shuffle。如果你在 SQL 里写了/*+ REPARTITION(1000) */，但后面又有一个group by，那个group by的 Shuffle 分区数还是由spark.sql.shuffle.partitions决定，不是 1000。hint 只影响它所在的那一层。

配置给完了，下面验证它到底有没有生效。

4. 验证请求与成功结果：用 Spark UI 和 explain 确认并行度真的变了

配置写完不代表生效，必须验证。我一般用三个手段：explain()看物理计划、Spark UI 看 Task 数、日志看 Shuffle 读写。

第一步，用 explain 看物理计划。

val df = spark.sql(""" SELECT /*+ REPARTITION(500, city) */ city, count(*) as cnt FROM orders GROUP BY city """) df.explain("formatted")

重点看输出里有没有Exchange RoundRobinPartitioning(500)或Exchange hashpartitioning(city, 500)。如果看到的是Exchange hashpartitioning(city, 200)，说明 hint 没生效，检查 Spark 版本是否 ≥ 2.4。

我实测的一个输出片段：

== Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- HashAggregate(keys=[city#10], functions=[count(1)]) +- Exchange hashpartitioning(city#10, 500) +- HashAggregate(keys=[city#10], functions=[partial_count(1)]) +- Exchange RoundRobinPartitioning(500) +- FileScan parquet ...

看到两个Exchange，第一个是RoundRobinPartitioning(500)，说明我的 hint 生效了；第二个是hashpartitioning(city, 500)，这是group by自己的 Shuffle，分区数也是 500，因为我同时设了spark.sql.shuffle.partitions=500。

第二步，Spark UI 看 Task 数。

作业跑起来后，打开 Spark UI（默认 4040 端口），进到 Stages 页面。找到对应的 Stage，看Tasks那一列。如果配置生效，Task 数应该等于你设的分区数。

具体路径：http://<driver-host>:4040→Stages→ 点进目标 Stage → 看Summary Metrics里的Total Tasks。

我一般会对比调优前后的两个 Stage：

指标	调优前	调优后
Total Tasks	200	2560
Duration	40 min	12 min
Shuffle Read	480 GB	480 GB
Shuffle Write	460 GB	470 GB
Max Task Duration	38 min	3 min

注意Max Task Duration这个指标，它最能反映倾斜。如果最大值远大于中位数，说明有长尾 Task，光加分区数没用，得按列重分区。

第三步，看日志确认 Shuffle 分区数。

在 executor 日志里搜shuffle，能看到类似：

INFO ShuffleBlockFetcherIterator: Getting 2560 non-empty blocks

或者 driver 日志里：

INFO DAGScheduler: Submitting 2560 missing tasks from ShuffleMapStage 3

2560就是实际的分区数。

第四步，用 TaoToken 辅助分析。如果你不想手动翻 UI，可以把 Stage 的 JSON 指标丢给模型，让它帮你判断瓶颈。比如：

import requests, os, json stage_metrics = { "total_tasks": 2560, "max_task_duration_ms": 180000, "median_task_duration_ms": 45000, "shuffle_read_bytes": 515396075520, "shuffle_write_bytes": 504658657280 } resp = requests.post( f"{os.environ['TAOTOKEN_BASE_URL']}/v1/chat/completions", headers={"Authorization": f"Bearer {os.environ['TAOTOKEN_API_KEY']}"}, json={ "model": os.environ.get("TAOTOKEN_MODEL", "gpt-4o"), "messages": [{ "role": "user", "content": f"分析这个 Spark Stage 指标，判断是否存在数据倾斜，给出重分区建议：{json.dumps(stage_metrics)}" }] } ) print(resp.json()["choices"][0]["message"]["content"])

返回会告诉你 max/median 比值超过 4 倍就存在倾斜，建议用repartition(n, key)或者加盐。

成功结果的判断标准：

1. 物理计划里出现预期的Exchange和分区数
2. Spark UI 的 Task 数等于配置值
3. Max Task Duration / Median Task Duration < 3
4. 总耗时下降，且 Shuffle 读写没有暴涨

如果这四条都满足，说明重分区调优成功。如果 Task 数对了但耗时没降，大概率是 Shuffle 开销吃掉了收益，需要减少分区数或者换coalesce。

5. 本篇常见错排查：401、local proxy failed、reading choices、OAuth 报错对照

调优过程中报错是常态，我把最常见的几类整理出来，对照着查。

错误一：401 Unauthorized

{"error": {"message": "Invalid API key", "type": "invalid_request_error"}}

原因：Key 没设、设错、或者环境变量没传到 executor。

排查步骤：

# 1. 确认 driver 上 Key 存在 echo $TAOTOKEN_API_KEY # 2. 确认 executor 也能拿到（关键！） spark-submit --conf spark.executorEnv.TAOTOKEN_API_KEY=$TAOTOKEN_API_KEY ...

很多人只在 driver 设了环境变量，executor 上取不到，就会 401。正确做法是用spark.executorEnv.前缀显式传递。

错误二：local proxy failed

java.net.ConnectException: local proxy failed: Connection refused

这个报错通常出现在你配置了本地网络代理，但代理服务没启动。注意：TaoToken 的调用不需要任何本地代理，直接走https://taotoken.net/api就行。如果你环境里有代理配置，检查：

echo $http_proxy echo $https_proxy

如果有值，且代理不可用，unset 掉：

unset http_proxy https_proxy

或者在 Spark 配置里排除：

spark.executor.extraJavaOptions=-Dhttp.proxyHost= -Dhttp.proxyPort= -Dhttps.proxyHost= -Dhttps.proxyPort=

错误三：reading choices 相关报错

java.lang.NullPointerException: reading choices

或者

com.fasterxml.jackson.databind.JsonMappingException: reading choices

这是解析模型返回 JSON 时，choices字段为空或结构不对。常见原因：

1. Model ID 写错，返回了错误信息而不是正常响应
2. 请求体格式不对，比如messages写成了message
3. 返回被截断，max_tokens设太小

排查：

curl -X POST https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{"model":"gpt-4o","messages":[{"role":"user","content":"test"}]}' | jq .

用jq看完整返回，确认choices数组存在且有内容。

错误四：OAuth 相关报错

OAuth token expired

或者

invalid_grant: token has expired

如果你用的是需要 OAuth 的模型接入方式，token 过期会导致这个错。TaoToken 的 API Key 方式不涉及 OAuth，如果你看到这个报错，说明代码里混用了其他 SDK 的鉴权逻辑。检查你的 HTTP 客户端是不是自动加了Authorization: Bearer之外的头。

错误五：hint 不生效

这个不报错，但结果不对。检查清单：

1. Spark 版本 ≥ 2.4（spark-submit --version）
2. hint 写在SELECT后面，不是FROM后面
3. 语法正确：/*+ REPARTITION(100) */，注意是/*+不是/*
4. 用explain()确认物理计划

错误六：Shuffle 失败，fetch failed

org.apache.spark.shuffle.FetchFailedException

分区数设太大，Shuffle 文件太多，或者磁盘满了。解决：

• 减少分区数
• 增大spark.shuffle.file.buffer
• 检查spark.local.dir磁盘空间

错误七：AQE 不生效

spark.sql.adaptive.enabled=true

设了但没效果，检查：

1. Spark 版本 ≥ 3.0（AQE 在 3.0 才正式 GA）
2. spark.sql.adaptive.coalescePartitions.enabled=true也要开
3. 看日志有没有AdaptiveSparkPlan isFinalPlan=true

错误八：Cline MCP 或 Codex auth.json 配置问题

如果你在调优过程中用 Cline 的 MCP 或者 Codex 的auth.json来辅助，三件套必须写全：

{ "base_url": "https://taotoken.net/api", "api_key": "sk-xxxxxxxx", "model": "gpt-4o" }

缺任何一个都会报鉴权或模型找不到的错。auth.json路径一般在~/.codex/auth.json，Cline 的配置在 VS Code 设置里。

排查完这些，基本能覆盖 90% 的问题。剩下的看日志关键字，grep -i error先定位。

6. 语义一致 CTA：把调优验证固定到统一通道

调优这件事，最怕的是"这次跑通了，下次换个环境又不行"。我的做法是把所有验证步骤都收敛到 TaoToken 统一通道上，Base URL、Key、Model ID 三件套固定，这样无论换集群、换版本，只要三件套不变，验证逻辑就能复现。

具体来说，我建议你按用途分流：

排障和接入类，比如 401、local proxy failed、reading choices 这些报错，直接去 API Keys 页面确认 Key 状态，再去接入文档对照配置：

• API Keys：https://taotoken.net/api-keys
• 接入文档：https://taotoken.net/doc

验证模型类，比如你想确认某个 Model ID 能不能用、返回格式对不对，用模型对话页面直接测：

• 模型对话：https://taotoken.net/chat

长期编码和 Agent 类，比如你每天都要跑 Spark 调优、让模型分析执行计划，用 Coding Plan 更划算：

• Coding Plan：https://taotoken.net/coding-plan

控制台看用量和调用记录：

• Console：https://taotoken.net/console

Claude Code 相关，如果你用 Claude Code 做 Spark 代码辅助：

• ClaudeCodeAnthropic：https://taotoken.net/claude-code

最后给一个我常用的收尾操作：把调优前后的物理计划和 Stage 指标存成 JSON，用模型做一次对比分析，确认并行度变化符合预期。这一步做完，整个重分区调优就算闭环了。

import json, requests, os before = {"partitions": 200, "duration_min": 40, "max_task_min": 38} after = {"partitions": 2560, "duration_min": 12, "max_task_min": 3} resp = requests.post( f"{os.environ['TAOTOKEN_BASE_URL']}/v1/chat/completions", headers={"Authorization": f"Bearer {os.environ['TAOTOKEN_API_KEY']}"}, json={ "model": os.environ["TAOTOKEN_MODEL"], "messages": [{ "role": "user", "content": f"对比调优前后指标，判断是否达到预期，还有没有优化空间：before={json.dumps(before)}, after={json.dumps(after)}" }] } ) print(resp.json()["choices"][0]["message"]["content"])

跑完这一步，你会得到一份带具体建议的分析，比单纯看数字直观得多。调优不是一次性的，数据量在变、集群在变，定期用这套流程复验，才能保证并行度始终匹配实际负载。