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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_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_KEYsk-xxxxxxxx export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODELgpt-4oWindows 下用set或者写进系统环境变量。设完之后echo $TAOTOKEN_API_KEY确认一下。这里插一句不要把 Key 提交到 Git。我见过太多人把 Key 写进spark-defaults.conf然后推到公共仓库。正确做法是用--conf spark.executorEnv.TAOTOKEN_API_KEYxxx在提交时注入或者用密钥管理服务。前置准备做完下面进入真正的配置环节。我会给你可以直接复制的 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.partitions800 # 自适应执行Spark 3.x 强烈建议开启 spark.sql.adaptive.enabledtrue spark.sql.adaptive.coalescePartitions.enabledtrue spark.sql.adaptive.skewJoin.enabledtrue # 单分区目标大小AQE 合并分区时参考 spark.sql.adaptive.advisoryPartitionSizeInBytes256MB # 读文件时的最大分区字节数 spark.sql.files.maxPartitionBytes128MB # 动态资源分配 spark.dynamicAllocation.enabledtrue spark.dynamicAllocation.minExecutors5 spark.dynamicAllocation.maxExecutors100如果用 Spark Submit 提交命令行参数spark-submit \ --master yarn \ --deploy-mode cluster \ --conf spark.sql.shuffle.partitions800 \ --conf spark.sql.adaptive.enabledtrue \ --conf spark.sql.adaptive.coalescePartitions.enabledtrue \ --conf spark.sql.adaptive.advisoryPartitionSizeInBytes256MB \ --conf spark.executorEnv.TAOTOKEN_API_KEY$TAOTOKEN_API_KEY \ --conf spark.executorEnv.TAOTOKEN_BASE_URLhttps://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 2560byCores 100 * 3 300结果 min(max(2560, 300), 5120) 2560所以设spark.sql.shuffle.partitions2560。但如果你开了 AQE其实可以设大一点让 AQE 自动合并spark.sql.shuffle.partitions4000 spark.sql.adaptive.coalescePartitions.enabledtrue spark.sql.adaptive.advisoryPartitionSizeInBytes256MBAQE 会在运行时把小的分区合并到 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 isFinalPlanfalse - 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.partitions500。第二步Spark UI 看 Task 数。作业跑起来后打开 Spark UI默认 4040 端口进到 Stages 页面。找到对应的 Stage看Tasks那一列。如果配置生效Task 数应该等于你设的分区数。具体路径http://driver-host:4040→Stages→ 点进目标 Stage → 看Summary Metrics里的Total Tasks。我一般会对比调优前后的两个 Stage指标调优前调优后Total Tasks2002560Duration40 min12 minShuffle Read480 GB480 GBShuffle Write460 GB470 GBMax Task Duration38 min3 min注意Max Task Duration这个指标它最能反映倾斜。如果最大值远大于中位数说明有长尾 Task光加分区数没用得按列重分区。第三步看日志确认 Shuffle 分区数。在 executor 日志里搜shuffle能看到类似INFO ShuffleBlockFetcherIterator: Getting 2560 non-empty blocks或者 driver 日志里INFO DAGScheduler: Submitting 2560 missing tasks from ShuffleMapStage 32560就是实际的分区数。第四步用 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: fBearer {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)或者加盐。成功结果的判断标准物理计划里出现预期的Exchange和分区数Spark UI 的 Task 数等于配置值Max Task Duration / Median Task Duration 3总耗时下降且 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 failedjava.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字段为空或结构不对。常见原因Model ID 写错返回了错误信息而不是正常响应请求体格式不对比如messages写成了message返回被截断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 不生效这个不报错但结果不对。检查清单Spark 版本 ≥ 2.4spark-submit --versionhint 写在SELECT后面不是FROM后面语法正确/* REPARTITION(100) */注意是/*不是/*用explain()确认物理计划错误六Shuffle 失败fetch failedorg.apache.spark.shuffle.FetchFailedException分区数设太大Shuffle 文件太多或者磁盘满了。解决减少分区数增大spark.shuffle.file.buffer检查spark.local.dir磁盘空间错误七AQE 不生效spark.sql.adaptive.enabledtrue设了但没效果检查Spark 版本 ≥ 3.0AQE 在 3.0 才正式 GAspark.sql.adaptive.coalescePartitions.enabledtrue也要开看日志有没有AdaptiveSparkPlan isFinalPlantrue错误八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.jsonCline 的配置在 VS Code 设置里。排查完这些基本能覆盖 90% 的问题。剩下的看日志关键字grep -i error先定位。6. 语义一致 CTA把调优验证固定到统一通道调优这件事最怕的是这次跑通了下次换个环境又不行。我的做法是把所有验证步骤都收敛到 TaoToken 统一通道上Base URL、Key、Model ID 三件套固定这样无论换集群、换版本只要三件套不变验证逻辑就能复现。具体来说我建议你按用途分流排障和接入类比如 401、local proxy failed、reading choices 这些报错直接去 API Keys 页面确认 Key 状态再去接入文档对照配置API Keyshttps://taotoken.net/api-keys接入文档https://taotoken.net/doc验证模型类比如你想确认某个 Model ID 能不能用、返回格式对不对用模型对话页面直接测模型对话https://taotoken.net/chat长期编码和 Agent 类比如你每天都要跑 Spark 调优、让模型分析执行计划用 Coding Plan 更划算Coding Planhttps://taotoken.net/coding-plan控制台看用量和调用记录Consolehttps://taotoken.net/consoleClaude Code 相关如果你用 Claude Code 做 Spark 代码辅助ClaudeCodeAnthropichttps://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: fBearer {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])跑完这一步你会得到一份带具体建议的分析比单纯看数字直观得多。调优不是一次性的数据量在变、集群在变定期用这套流程复验才能保证并行度始终匹配实际负载。