最近有好几个团队过来问我同一个问题手里的 Spark 集群跑着一堆离线数据现在想把大模型接进来做文本分类、实体抽取、情感打分这类活儿到底该怎么做有的同学上来就打算在 UDF 里把模型 load 进去结果集群直接 OOM有的走 HTTP 逐条调大模型接口跑了一晚上一看进度条才走了几百万条。今天就把我在多个项目里摸爬滚打总结出来的方案和细节整理出来覆盖方案选型、代码写法、资源规划和最容易被坑坏的那些细节。先说明白这篇文章适合谁看你已经会写 PySpark跑过简单的 join 或 groupBy现在需要把大模型推理任务和大数据 pipeline 结合在一起或者是刚接手一个所谓“LLM Spark”项目想知道别人是怎么做的。全文以实操为主尽量不堆概念。1. 为什么偏偏要在 Spark 里调用大模型1.1 把“存量数据拥抱大模型”这件事搬上生产先说场景。现在的数据团队手里普遍压着大量历史数据几亿条用户评论、合同扫描件转出来的文本、客服对话记录、商品描述甚至是一堆日志字段。过去这些数据要么洗完之后做统计报表要么用 TF-IDF 和词向量凑合出一个“语义相似度”但真正想要的理解型任务比如判断一段评论到底是“质量抱怨”还是“物流抱怨”传统方法做不动。大模型出来之后大家第一个想到的是让 LLM 干这个活。但几亿条数据不可能靠脚本一条条调接口也不可能全部塞进 Excel。这时候 Spark 作为离线批处理的事实标准自然而然就成了很多人眼里“调度这些数据”的首选。你只需要在 Spark 里写个 UDF把 DataFrame 的一部分字段传进去调用大模型把结果作为新列写出来逻辑上非常顺。但逻辑顺不等于跑得顺。我见过不少项目就在这一步开始炸模型参数几百 G每个 Executor 都 load 一遍内存直接翻车或者从 Hive 读一张表每条 row 都发起一次 HTTP 请求网络 IO 成了瓶颈跑了好几天都跑不完。所以先搞清楚 Spark 里调大模型到底有哪些靠谱套路比直接写代码更重要。1.2 先理解 UDF 在集群里的真实运行位置很多人对 Spark UDF 有一个误解以为 UDF 是在 Driver 上统一执行的其实完全不是。Spark 跑批任务时数据被切成多个 partition分布在不同的 Executor 上每一个 partition 会被一个 Task 处理。你写的 UDF 会被序列化之后分发到每个 Executor然后由 Executor 里的 Python worker 进程逐行执行。这带来一个非常关键的影响如果 UDF 里要访问某个对象比如一个大模型、一个 HTTP 客户端、一个数据库连接池那么这个对象要么能跟着 UDF 一起被序列化分发要么就必须在 Executor 的 worker 进程里独立初始化。理解了这一点后面很多坑就都能解释了为什么模型不能直接在 UDF 里 load因为每个 Executor 都要 load 一份你有 100 个 Executor就得占 100 份显存或内存为什么很多人的 HTTP 请求特别慢因为每个 Task 可能在不同时间启动连接没法复用甚至每次请求都要重新建立 TCP 连接。没有这种全局视角后面调优基本靠瞎试。2. 四种主流接入方案选型对比与推荐场景2.1 方案一在 UDF 里直接加载本地或内网模型这是新手最容易想出来的方案既然要调大模型那直接把模型放到集群的每台机器上在 UDF 里用transformers库加载然后逐条推理。听起来很美好实际只适合小模型。具体的做法是一个 map 类型的 UDFfrom pyspark.sql.functions import udf from pyspark.sql.types import StringType udf(StringType()) def local_model_predict(text): from transformers import pipeline # 注意如果在这里加载模型每条 task 第一次执行时都会触发加载 classifier pipeline(sentiment-analysis, model/path/to/model) result classifier(text, truncationTrue)[0] return result[label]这种写法最大的问题在于模型加载时机。如果 pipeline 在 UDF 内部初始化意味着每个 Python worker 的每个 task 第一次调用时都可能触发一次模型加载。即使模型只有几百 MB100 个 Executor 同时加载也会把磁盘 IO 和内存打满。更别提真正的 LLM动辄 7B、13B 参数光权重就要十几 GB 甚至几十 GB。所以这个方案我只建议用在两类场景一是模型很小比如几百万参数的 MiniLM、BERT-base 这类二是你只需要在单个 Executor 上做一次性处理而不是全集群高并发推理。如果非要在大模型场景里用也可以把模型放在共享文件系统比如 HDFS 或对象存储让 Executor 从远端加载但启动时间会非常感人而且多个 worker 同时拉模型文件可能把带宽占满。2.2 方案二逐条或小批量调用在线 LLM API这是目前生产环境里最常见的做法。大模型统一以一个在线服务的形式部署在内网Spark 这边不去管模型本身只管发 HTTP 请求、拿结果。这个在线服务可以是外部厂商的 API也可以是团队自己用 vLLM、TGI、SGLang 之类框架部署的模型服务。最简单的实现长这样import requests from pyspark.sql.functions import udf from pyspark.sql.types import StringType def call_llm(text): resp requests.post( http://llm-service:8000/v1/completions, json{prompt: text, max_tokens: 128}, timeout30, ) resp.raise_for_status() return resp.json()[choices][0][text] llm_udf udf(call_llm, StringType()) df.withColumn(llm_result, llm_udf(df[content]))逻辑很简单但性能一言难尽。每条数据一次请求如果数据量是一亿条哪怕每次请求只要 200 毫秒单并发也要跑两百多天。再加上网络开销、限流、超时这个方案不经过任何优化就跑全量数据基本等于自杀。所以这个方案必须搭配批量优化。我们要做的核心是两件事一是尽量多利用模型服务的并发能力二是减少 HTTP 请求的次数或提高每次请求的处理量。下面第三章我会详细讲怎么用 pandas UDF 和 Iterator 风格实现真正的批量调用。2.3 方案三Spark 调度 外部离线推理服务如果说方案二是 Spark 直接跟在线服务同步交互那方案三是把一个完整的异步离线推理链路搭起来。Spark 的角色变成“数据组织和结果回收方”真正的大模型推理发生在一个独立的服务集群里。大致流程是这样的Spark 批量读数据筛选出需要推理的字段把数据写入消息队列Kafka、Pulsar或中间表一个常驻的推理worker从队列里拿数据调用部署好的大模型服务做推理再把结果写回结果表Spark 那边可以用流式任务或者定时批任务把结果读回来跟原始表做 join得到最终带推理结果的数据。这个方案的优点是 Spark 不再直接依赖大模型服务的实时性能和稳定性推过去的数据可以慢慢消化模型服务即使重启、扩容也不影响 Spark 主任务。缺点是架构复杂度明显上升你需要额外维护队列和推理 worker。如果你们团队已经有现成的推理平台或者大模型服务经常抖动我建议优先考虑这个方案如果只是临时跑一次数据就太重了。2.4 方案四引入 Ray 或其他分布式推理引擎做协同比方案三更激进一点的做法是直接用 Ray 这类通用分布式计算框架管理大模型推理Spark 和 Ray 各自负责自己擅长的部分。比如 Spark 负责从 Hive 抽数、做复杂的 SQL 清洗和 join然后你可以把要推理的数据转换成 Ray 的 object refRay 侧用多个 GPU actor 并行加载模型、做推理最后结果再转回 Spark DataFrame 做下游统计。这个方案适合的场景有两个特征一是模型推理量非常大二是你们团队的 GPU 资源本身已经用 Ray 或者类似框架在管理。如果只是为了一个批处理任务专门搭一套 Ray 集群代价会比较高。而且 Spark 和 Ray 之间的数据打通没那么自然通常需要走 Parquet 文件或者 Redis 之类的中间存储增加了延迟。2.5 各方案选型对比下表是我个人在项目里做选型时的参考框架方案优点缺点适合场景UDF 加载本地小模型部署简单无外部依赖大模型内存爆炸加载慢百 MB 级小模型低并发逐条/小批量调用在线 API逻辑简单模型可集中管理性能差易被限流数据量小、临时验证Spark 外部离线推理服务解耦好模型服务可独立扩展组件多链路长大规模常态化推理任务Spark Ray 分布式推理推理性能上限高架构重运维复杂GPU 密集型、高吞吐需求从我自己的实践来看大多数团队的选型会落在方案二和方案三之间。临时跑一版用方案二常态化跑批就要升级成方案三。方案四适合已经有该基础设施的团队不建议现搭。3. 实战演练给上亿条评论做大模型情感分类3.1 设定一个具体任务我拿一个真实案例来演示。假设我们有一张 Hive 表app_comments里面存了电商平台的用户评论核心字段是comment_id和comment_text。现在要做两件事给每条评论打一个情感标签positive / negative / neutral再提取一个话题标签比如“物流”、“质量”、“价格”、“服务”。全量数据大约 1 亿条我们部署了一个内网模型服务兼容 OpenAI 的 Chat 接口风格QPS 上限约 200。这个任务如果按方案二直接逐条调显然不行。就算按 200 QPS 跑满1 亿条也要 5 万秒差不多 14 个小时。但实际网络、限流、重试损耗一叠加跑 24 小时都算运气好。所以这里我用 pandas UDF 加 Iterator 模式来写尽可能把批量优势发挥出来。3.2 先封装一个独立的 Python 调用函数不管 Spark 那边怎么包最关键的是先把“调用大模型”这件事做成一个纯净的 Python 函数。这样方便本地调试也方便后面放到 Spark 的不同封装模式里。这里用一个 OpenAI 兼容接口的写法假设请求参数是model、messages、temperature那些。如果你们有自己的协议替换成requests.post的 body 就行。import requests import time from typing import List, Dict API_URL http://llm-service:8000/v1/chat/completions API_KEY internal-token def llm_chat(prompt: str, max_tokens: int 256) - str: payload { model: qwen2.5-7b-instruct, messages: [ {role: system, content: 你是一个文本分析助手只输出 JSON。}, {role: user, content: prompt}, ], temperature: 0.0, max_tokens: max_tokens, } resp requests.post( API_URL, jsonpayload, headers{Authorization: fBearer {API_KEY}}, timeout60, ) resp.raise_for_status() return resp.json()[choices][0][message][content]这个函数我没加任何重试逻辑因为重试要放到批量层去做不然单个函数里写死退避逻辑会拖慢整体吞吐。这里高亮一个关键点超时必须设不要用默认的无限等待。大模型服务有时会因为排队过长导致单个请求十几秒不回如果不设超时整个 Spark Task 就会一直挂着最后变成一副“集群活着但任务不动”的诡异画面。然后再加一个解析函数大模型输出并不总是稳定的 JSON有时会带解释文字有时直接给你一段 Markdown。所以我一般会写一个容错解析提取 JSON 片段再解析解析失败就返回一个默认结果。import json import re def parse_llm_json(text: str) - Dict: text text.strip() # 去掉可能的 json 包装 text re.sub(r^(?:json)?|$, , text, flagsre.MULTILINE).strip() try: return json.loads(text) except json.JSONDecodeError: # 尝试找到第一个 { 到最后一个 } start text.find({) end text.rfind(}) if start ! -1 and end ! -1 and end start: try: return json.loads(text[start:end1]) except json.JSONDecodeError: pass return {sentiment: unknown, topic: unknown}为什么这么写因为实际跑过就知道大模型偶尔会在 JSON 前后加解释或者中途截断。你要是直接把json.loads的结果当成最终结果出错率不高但一亿条数据哪怕 1% 的脏数据也是 100 万条要重跑很痛。3.3 用 pandas UDF 包装批量推理接下来是 Spark 部分。我不建议用普通 UDF因为逐行调用llm_chat的话每行都是一次完整的 HTTP 往返延迟被完全暴露。更合理的是用 pandas UDF 的 Iterator 模式每次处理一批数据这样可以在批内复用逻辑、控制并发甚至可以把一批文本拼接成一次请求。基础的 Iterator 写法是这样的from pyspark.sql.functions import pandas_udf import pandas as pd pandas_udf(string) def sentiment_topic_batch(iterator): for batch in iterator: # batch 是一个 pandas Series代表一批 comment_text texts batch.tolist() results [] for text in texts: prompt ( 请对以下用户评论进行情感和话题分析 返回 JSON格式为 {\sentiment\: \positive|negative|neutral\, \topic\: \物流|质量|价格|服务|其他\}。\n f评论{text} ) raw llm_chat(prompt) parsed parse_llm_json(raw) # 把 dict 转成字符串后面统一解析 results.append(json.dumps(parsed, ensure_asciiFalse)) yield pd.Series(results)然后在主代码里这样调用df spark.sql(SELECT comment_id, comment_text FROM app_comments WHERE comment_text IS NOT NULL) result_df df.withColumn(llm_result, sentiment_topic_batch(df[comment_text]))跑起来之后你会发现它比逐条 UDF 快但还没达到理想状态。因为这里的llm_chat仍然是每条文本一个请求只是把 Spark Task 的序列化开销和行处理开销省掉了一部分。真正的吞吐瓶颈还是在大模型服务的 QPS 上。所以如果要进一步提速有两种常见做法一种是把文本改成批量提示词让模型一次处理多个文本减少请求数另一种是在 Executor 里做并发请求一次批数据拆成几个线程同时调接口充分利用模型服务的 QPS。我下面把两种都说一下。3.4 批量提示词与并发请求的取舍批量提示词的思路是把这一批 50 条评论拼成一个 JSON 数组塞进一个 prompt让模型一次性返回 50 条结果。def llm_chat_batch(texts: List[str]) - List[str]: body json.dumps([{id: i, text: t} for i, t in enumerate(texts)], ensure_asciiFalse) prompt ( 请逐条分析以下评论数组中的每条评论返回一个 JSON 数组 每个元素是 {\\\sentiment\\\: ..., \\\topic\\\: ...}下标保持对应。\n f数组{body} ) raw llm_chat(prompt, max_tokens1024) try: parsed json.loads(raw) # 按长度对齐不完整时用 unknown 补 return [json.dumps(x, ensure_asciiFalse) for x in parsed[:len(texts)]] except json.JSONDecodeError: return [json.dumps({sentiment: unknown, topic: unknown}) for _ in texts]这种做法能大幅降低请求次数比如原来 50 条要 50 个请求现在只需要 1 个。但风险在于模型可能不严格对齐下标或者输出被截断导致整个 batch 解析失败。一旦失败这一批全部要重来反而更慢。所以我把 batch 大小控制在 2050 之间并且解析容错做得比较强。另一种做法是并发请求。在 pandas UDF 里对一批 100 个文本用 ThreadPoolExecutor 同时发 510 个请求等待全部返回。这样可以比较充分地利用模型服务的 QPS而且因为是在 Executor 端并发不会受 Spark 单 Task 串行限制。from concurrent.futures import ThreadPoolExecutor, as_completed def call_llm_with_limited_concurrency(texts: List[str], max_workers: int 8) - List[str]: results [None] * len(texts) with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map { executor.submit(llm_chat, build_prompt(texts[i])): i for i in range(len(texts)) } for future in as_completed(future_map): idx future_map[future] try: raw future.result() results[idx] json.dumps(parse_llm_json(raw), ensure_asciiFalse) except Exception: results[idx] json.dumps({sentiment: unknown, topic: unknown}) return results这里有个非常重要的事并发数不要拍脑袋定。如果模型服务 QPS 上限是 200平均每个请求耗时 300ms那么单个 worker 的并发上限差不多是 60因为 1 秒 / 0.3 秒 × 20 QPS 其实这个数字要根据模型吞吐算。在实际项目里更稳妥的办法是拿到模型服务的压测数据再定。第一版可以先保守一点并发设 48跑五分钟看看服务端监控的 QPS 和延迟再逐步上调。完整的 pandas UDF 代码就变成了pandas_udf(string) def sentiment_topic_batch(iterator): for batch in iterator: texts batch.tolist() # 20 条一组拼接或直接并发逐条 yield pd.Series(call_llm_with_limited_concurrency(texts, max_workers8))3.5 全量任务怎么跑从抽样到扩容我不建议拿 1 亿条直接开跑。第一次调试时先读 1 万条数据确认结果格式正确、服务端没有报错再逐步放大。比如先跑 100 万条看单 Task 的平均耗时推算全量所需时间再决定要不要加资源。假设 100 万条数据跑了 3 分钟那 1 亿条就差不多要 300 分钟5 个小时。如果你觉得太慢可以加大spark.sql.shuffle.partitions没用关键是增加 Executor 数量和并发度。比如从 50 个 Executor 加到 200 个单 Task 处理的数据量不变但并行度上去了总时间理论上是原来的四分之一。但也要注意加太多 Executor 不一定是好事。大模型服务端如果 QPS 扛不住加再多 Spark Executor 也只会看到大量超时和重试。正确的做法是先确认服务端瓶颈在哪。我常用的思路是把 Spark UI 里的 Task 耗时和服务端监控对照看如果 Task 平均耗时接近请求平均耗时说明瓶颈在网络或服务端加 Spark 资源没用如果 Task 里有大量时间花在处理和排队上那确实需要提高并行度。4. 我踩过的那些坑序列化、超时、限流与资源配置4.1 为什么请求函数经常传不进去很多人写完 UDF 提交任务直接在 Driver 端报错类似Could not serialize object: TypeError: cannot pickle xxx object。原因是你把某个无法被 pickle 的对象定义在了 UDF 所在的作用域里比如在类里初始化了一个 HTTP 连接池或者把一个大模型的pipeline对象定义成全局变量。PySpark 在分发 UDF 时会把相关对象通过 pickle 序列化后发给 Executor。像requests.Session虽然勉强能 pickle但序列化之后连接池状态会丢一些模型对象、GPU 句柄根本无法序列化。解决办法就是不要在 UDF 对象里直接持有这些资源而是把它们的初始化逻辑放到 Executor 内部惰性执行。比如在 UDF 里判断全局变量_client是不是 None是就现场建一个_client None def get_client(): global _client if _client is None: _client LLMClient() return _client这样 UDF 被序列化时只发送了一个函数引用真正重的客户端对象在每个 Executor 的 worker 进程里各自实例化一次。这也是为什么很多人说“在 mapPartitions 里初始化一次连接池”才是正确姿势。4.2 Executor 核数、内存与并行度怎么配Spark 调大模型推理和普通的 ETL 任务不一样CPU 和内存都不是绝对瓶颈网络 IO 和模型服务端 QPS 才是。但这不代表资源随便配。我见过不少同事开 100 个 Executor、每个 Executor 8 核结果大模型服务被打挂任务疯狂重试。反过来如果你只开 5 个 Executor每个核数也不高那又白白浪费了模型服务的吞吐能力。一个比较稳的先手配置是先限制spark.executor.cores2每个 Executor 内存 48G这样单个 Executor 的并发请求数基本在 24取决于你在 UDF 里开的线程数。跑 5 分钟看一下模型服务监控的 QPS 是否接近上限如果离上限还很远再增加 Executor 数量或者把executor.cores调高。注意executor.cores调高不等于 Spark 里每个 Task 就会自动并发发起 HTTP 请求除非你在 pandas UDF 内部自己开了线程。另外有个容易被忽略的点spark.sql.execution.arrow.enabled最好保持默认开启因为 pandas UDF 依赖 Arrow。如果你的 Spark 版本比较老Arrow 性能有问题可以升级 Spark 或 pyarrow 版本而不是关掉 Arrow 硬跑。4.3 超时、限流、熔断一个都不能少在线模型服务不像你本地写的函数它受队列长度、GPU 显存、batch size 等因素影响响应时间起伏很大。所以请求必须设置超时最好还能区分连接超时和读超时。我常用的请求配置是timeout(10, 120)意思是连接超时 10 秒读超时 120 秒。大模型长文本生成可能确实需要较长时间所以读超时不能太短。对于限流模型服务一般会返回 429 或者 503。429 是请求太频繁503 是服务暂时不可用。处理方式不同429 可以稍微退避后重试503 要等更长时间甚至放弃这一批。我会在批量函数里加一个简单的重试逻辑最多重试 3 次退避时间按 1 秒、2 秒、4 秒递增import time def call_with_retry(prompt, retries3): for attempt in range(retries): try: return llm_chat(prompt) except requests.exceptions.HTTPError as e: if e.response.status_code in (429, 503) and attempt retries - 1: time.sleep(2 ** attempt) continue raise except requests.exceptions.ReadTimeout: if attempt retries - 1: time.sleep(1) continue raise这里有一个容易犯错的地方不要对每条请求都在内部重试因为重试会叠加在批量并发之上可能瞬间造成更大的请求洪峰。正确做法是把重试次数压到很低12 次并且配合并发数上限来控制总体 QPS。如果错误率持续超过 5%应该停下来查模型服务状态而不是让重试逻辑硬扛。再补一个熔断思路在做超大任务之前先往模型服务发 50 个测试请求统计成功率。如果成功率低于 90%就不要启动 Spark 全量任务先解决模型服务的问题。这个前置检查写成一个单独的 Python 脚本不进 Spark用来给平台做发布前检查非常有用。4.4 token 长度、输出截断与结果校验大模型服务的输入输出都有 token 上限。文本如果太长接口会直接报错也可能自动截断。我一般在调用前做一次文本长度截断比如保留前 1000 个字符因为情感分类和话题识别对文本开头的信息最敏感。这样做有两个好处一是减少 token 消耗变相降低成本二是避免因为超长文本导致请求失败。输出截断是另一个问题。max_tokens设太小模型输出到一半被切断JSON 不完整解析就会失败。所以解析函数要非常健壮解析不了就返回 unknown。千万不要让一个坏 JSON 导致整个 Spark Task 失败那样重跑的成本高得多。我还会做一道结果完整性校验跑完 Spark 任务后统计llm_result里包含unknown的比例。如果 unknown 比例超过 1%说明模型输出质量有问题需要检查是提示词还是服务端配置出了问题。这个比例可以作为质量监控指标写进平台报表。4.5 失败重试的粒度批内重试优于批级重试很多人写 pandas UDF 时把整个 batch 包在 try except 里一旦 batch 里有一条请求超时整个 batch 都标记失败然后 Spark 重跑这个分区。这个做法非常浪费。因为一个 batch 里其他 99 条可能都已经成功返回了只是某一条超时你重跑整个 batch等于把 99 条成功的请求又打了一遍白白浪费资源还可能压垮模型服务。我建议在 batch 内部做单条重试。也就是上面call_llm_with_limited_concurrency里给每条请求单独 try except异常时单独重试。即使这条最终失败也只影响这一条输出 unknown 或默认值不影响任务整体跑完。后续如果对 unknown 结果有要求可以再单独捞出来重跑一次。这里也解释一下为什么建议把结果解析函数写得尽量宽松你的目标是让全量任务稳定跑完而不是追求单条结果完美。稳定跑完意味着任何异常都要被兜住绝不能因为某条脏数据让整个 Executor 崩溃。4.6 常见问题速查表现象可能原因排查与处理任务一启动就报序列化错误UDF 持有无法 pickle 的对象把客户端初始化改为惰性全局变量内存持续上涨直到 Executor 挂掉每条 Task 内 load 了大模型或连接池过多改为 mapPartitions 初始化控制连接池大小数据量很大但跑得极慢逐条调用 HTTP网络往返成为瓶颈改为批量提示词或并发请求任务卡住不动Spark UI 显示 Task 仍在运行请求没设超时模型服务长时间不返回设置 read timeout并加重试模型服务返回大量 429/503并发过高超过服务 QPS减少 Executor 或线程并发数加退避重试输出结果大量是 unknown提示词不合适或 max_tokens 太小调整提示词增大 max_tokens增加校验4.7 成本与性能估算先算账再开跑最后再分享一个经验大模型调用是有成本的不管是外部 API 按 token 计费还是内网 GPU 资源占用跑之前最好算一笔账。估算方式很简单假设单条评论平均 200 个 token输入输出加起来大约 400 token。如果批量提示词方案会稍微多一点但可以忽略。1 亿条就是 400 亿 token。外部 API 按百万 token 计费假设每百万 token 0.5 块总成本就是 2000 块。如果觉得贵就要考虑是否抽样子集、降低任务频率或者用小模型替代大模型。时间估算也有公式总请求数 / (并发数 × 单请求 QPS 贡献)。假设并发 8单请求 1 秒8 并发每秒 8 个请求1 亿条需要 12500 秒约 3.5 小时。这个估算结果可以帮你说服业务方接受调度时间也能帮你自己决定要不要加资源。5. 架构层面的一些个人体会踩过几次大坑之后我现在的习惯是把“在线”和“离线”分开。真正跑全量批处理的任务尽量不要让 Spark 直接对高频在线推理服务发请求因为两者生命周期不一样Spark 任务跑完就结束了在线服务却要常年稳定。如果 Spark 侧并发控制不好一次全量任务就可能把在线服务的资源打满影响线上业务。更稳的做法是把推理请求打到独立的离线模型服务池上或者在中间加一个削峰队列。另外prompt 的管理不要散落在代码里。我见过很多团队把 prompt 写死在 UDF 里后面想改一点措辞就要改代码重跑全量。更合理的做法是把 prompt 模板存到配置中心或 HDFS 上的一个配置表里Spark 任务启动时读取这样调 prompt 不用重新发布代码只需要重启任务。对于任务频率高、团队人员流动大的项目这个优化非常值。最后一点是监控。Spark 任务跑多久、失败多少条、unknown 比例多少、平均耗时多少这些指标最好都能汇总到团队已有的监控平台上。大模型推理跟普通 ETL 不一样它的结果质量波动很大今天 prompt 还能用明天换了个模型版本可能结果就变了。没有监控和对比后面排查问题会非常痛苦。这篇文章里提到的代码和方案都是我实际跑过之后沉淀下来的。如果你现在正要动手建议不要贪多第一版先用 pandas UDF 加并发请求把端到端链路跑通再逐步往异步队列或 Ray 方向演进。大模型在 Spark 里调用这件事最大的难点从来不是写代码而是把资源、并发、稳定性、成本这些细节都考虑周全。希望这篇分享能帮你少走几步弯路。