企业官网建设流程全解析

如果你在数据团队里待过一阵子，大概率会遇到这种需求：给几百万条文本打标签、抽取实体、做情感分析，或者把用户聊天记录批量生成模型训练样本。业务方开口就是“用大模型跑一下就行”，但真到动手阶段才发现，把大模型接进 Spark 里并没有想象中那么无脑。

最常见也最容易被吐槽的写法，是在 UDF 里直接requests.post调大模型 API。跑小数据量看不出问题，一旦上了千万级数据，连接打满、超时重试混乱、Executor 被拖垮，各种问题一起冒出来。这篇文章我想从工程视角把“Spark 调用大模型”这件事完整拆开：先说清楚哪些场景适合做、哪些场景其实不适合；然后对比几种主流的接入方案，给出选型建议；再放出可以直接改着用的 PySpark 代码，把并发、超时、重试、序列化这些硬骨头一个一个讲透。

适合读这篇文章的人，我默认是两类：一类是做数据平台、离线数仓、算法工程的开发，正在琢磨怎么把大模型能力批量落到数据管道里；另一类是准备数据开发或大数据架构面试的同学，因为这个问题这两年出现频率越来越高，面试官问的不是你会不会调 API，而是你能不能把分布式计算和大模型服务之间的资源、并发、容错关系讲明白。我会尽量把代码、参数、踩坑记录都放出来，文章里的路径都是我实际跑过的，照抄可以少走很多弯路。

1. 为什么会在 Spark 里调用大模型：场景与挑战

1.1 典型应用场景

先说清楚一件事：用 Spark 调大模型，本质上是“批量推理”，不是“在线推理”。在线推理要求毫秒级响应，那是后端服务的事，跟 Spark 没什么关系。Spark 的定位是离线批处理，所以凡是“有一批存量数据，想用大模型处理一遍”的场景，都适合拿到这里来讨论。

我实际见到过并且自己参与过的场景大概有这么几类：

• 文本批量打标与分类：比如把平台上积累的历史工单、评论、投诉内容统一做一次意图分类，量级通常在百万到千万条。用规则或小模型效果不好，用大模型跑一遍就能把标签体系更新一遍。
• 信息抽取与结构化：从非结构化文本里抽出时间、地点、金额、人物等实体，或者从长文档里抽取摘要。这个场景大模型比传统 NER 模型灵活得多，尤其适合字段不固定、规则经常变的抽取任务。
• 离线生成训练样本：用大模型给原始数据生成蒸馏样本、构造正负例、做数据增强。比如先用大模型生成一批高质量的问答对，再拿去微调小模型，这是现在很多团队降本增效的标准路径。
• 内容审核与风险识别：把待审核的数据批量过一遍大模型，输出风险等级、违规类别。这类任务往往要和规则引擎结合，Spark 负责把规则命中的部分再交给大模型做二次判断。
• 知识库向量化前置处理：虽然向量化通常用小模型做，但在构建 RAG 知识库时，很多团队会先用大模型对文档做清洗、切分、标题生成、摘要，再做 embedding。

这些场景有一个共同点：输入侧数据规模大，输出侧对实时性要求不高，但要求在合理时间内跑完，并且要能监控进度、控制成本、失败能重试。

1.2 核心矛盾与常见误区

Spark 调用大模型，本质上是在两个完全不同的系统之间做桥接。Spark 是分布式批处理框架，数据分布在很多 Executor 上并行算；大模型是外部服务，不管是公有云 API 还是私有化部署的推理服务，它有自己的并发上限和时延。两者之间最核心的矛盾有三个：

第一个是连接与并发模型不匹配。Spark 的并行度可以开到成百上千，一个 Executor 上可以同时跑很多 Task。如果你在每个 Task 里都新建一个 HTTP 连接去调 API，很快会把服务端连接池打满，也会触发限流。更麻烦的是，Python UDF 是逐行调用的，每一行都做一次完整的 HTTP 往返，连接无法复用，性能会非常难看。

第二个是资源模型的冲突。大模型推理需要 GPU，而 Spark 的 Executor 通常跑在 CPU 容器里。就算你用的是 Spark 3.4 之后支持 GPU 调度的版本，真正跑大模型推理时，GPU 显存和 Executor 堆内内存的管理也是两套逻辑。多数情况下，让 Spark 做数据准备和结果回填，把推理压力放在独立服务上，才是合理的分工。

第三个是容错哲学完全不同。Spark 对 Task 失败的态度是“重新算一遍”，某个分区失败了就重新执行该分区。但大模型调用是有外部副作用的：Token 已经消耗了，API 可能已经返回了结果只是网络超时了。如果你很天真地开启 Spark 重试机制，同一个 prompt 可能被重复发送很多次，账单会很难看。所以在 Spark 里调用大模型，重试逻辑必须自己控制，不能依赖 Spark 的 Task 重试。

很多人问我第一个问题就是：“直接写个 UDF 调一下不行吗？”行，数据量小确实行，我就这么干过。但“能跑”和“能用”是完全两码事。如果不控制并发、不复用连接、不做超时管理，百万级数据基本必挂。下面我会把不同方案摆出来对比，你就能明白为什么工程上最终都走向了“客户端池化 + 分区处理”这条路。

2. 方案选型对比：四种主流调用思路

2.1 方案A：UDF 直连 HTTP API

这是最容易上手、也最容易翻车的方案。直接在 PySpark 里注册一个 UDF，UDF 内部用requests或 OpenAI 兼容 SDK 调大模型 API，然后df.withColumn("result", call_llm_udf(col("text")))一把梭。

from pyspark.sql.functions import udf from pyspark.sql.types import StringType import requests def call_llm(text: str) -> str: resp = requests.post( "https://api.example.com/chat/completions", json={"model": "demo-model", "messages": [{"role": "user", "content": text}]}, timeout=30 ) data = resp.json() return data["choices"][0]["message"]["content"] call_llm_udf = udf(call_llm, StringType())

这个方案在小数据量验证时非常方便，几千条数据跑一遍完全没问题。但问题也很直接：

• 每行数据都新建一次 HTTP 连接，TCP 握手开销巨大，而且很多 SDK 内部会自己管理连接池，放在 UDF 里每行新建客户端，连接池根本来不及复用。
• Spark 默认的并行度如果比较高，会在同一瞬间发起大量并发请求，直接把服务端打挂，或者触发 429 限流。
• 超时和重试逻辑写起来很别扭。timeout一设，模型生成时间长一点就超时；不设超时，某个请求卡住，整个 Spark 任务可能挂在最后几个 Task 上，特别难受。

我的建议：这个方案只用来做连通性验证或百条级以下的试验，不要用于生产。如果你判断数据量短期内都不会超过几千条，那用这个方案也无可厚非，但最好还是提前预留改造空间。

2.2 方案B：mapPartitions + 客户端池化

这是目前工程上最主流、性价比最高的方案。核心思路很简单：不用 UDF 逐行调用，而是在每个分区内部初始化一次客户端，然后在这个分区内用并发线程池去调用大模型。

为什么用mapPartitions？因为 Spark 中数据是以分区为单位的，一个分区内的数据会在同一个 Executor 上顺序或并行处理。在分区级别做初始化，客户端只需要创建一次，连接可以复用；同时你可以在分区内部用线程池控制并发，把并发量稳定在一个服务端能接受的范围内。

这段逻辑用代码表达大致是这样的框架：

def process_partition(iterator): # 每个分区只会执行一次初始化 client = LLMClient(...) for row in iterator: yield do_call(client, row)

这里有几个容易踩的细节：如果你在 UDF 或者map里创建客户端，客户端会被序列化复制到每个 Task 里，要么报Task not serializable，要么每个 Task 都重新建连接，性能还不如方案A。而在mapPartitions里创建，客户端只属于当前分区所在的 Python 进程，一切都在本地，不涉及序列化问题。

我在后面的实战章节会给出完整的可运行实现，这里先不展开。

2.3 方案C：外部服务化 + 消息队列 + 结果回填

当数据量到了日均千万甚至亿级，或者大模型服务不稳定、需要独立扩缩容的时候，方案B也会遇到瓶颈。这时更稳妥的做法是把 Spark 作业拆成两个阶段：Spark 负责数据预处理和下发任务，独立推理服务负责消费和执行，结果再回填到表里。

具体的流程一般是这样的：

1. Spark 读取原始数据，清洗、裁剪、拼 prompt，把任务写入消息队列（Kafka 或 RocketMQ），消息体里带上业务主键 ID 和 prompt。
2. 独立的推理服务集群从消息队列消费任务，调用大模型 API 或本地推理服务，得到结果后写回结果表。
3. Spark 在第二阶段（或者另一个周期调度）扫描结果表，把有结果的记录和原始表关联，完成后续加工。

这个方案把 Spark 和大模型服务之间的耦合彻底解开了。推理服务可以按照模型吞吐独立扩缩容，队列可以在服务端抖动时起到削峰填谷的作用，Spark 不需要关心调用大模型的细节，只关心数据怎么写、怎么读。调用的吞吐上限从“Spark 分区并发数”变成了“消息队列的消费能力”，扩展性好了很多。

代价也很明显：架构复杂度增加了。你需要维护消息队列、消费服务、结果表的状态管理，还要处理消息丢失、重复消费、结果回填延迟等问题。我的判断是：如果你的数据量单次任务在百万到千万条量级，用方案B就够了；超过这个量级且要常态化跑，再考虑方案C。不要一上来就上消息队列，很多团队就是被自己的过度设计拖垮的。

2.4 方案D：本地 GPU 推理服务 + Spark 协同

这一两年本地模型部署工具成熟得很快，VLLM、TGI、Ollama 这些项目已经能把开源模型部署成 OpenAI 兼容的 HTTP 服务。于是有一种新的玩法：Spark 集群边上放一台或几台 GPU 服务器，用 VLLM 部署开源模型，Spark 在内网直接调用这个服务。

这个方案最大的优势是数据不出域。对于金融、医疗、政务这类对数据出境有严格要求的场景，公有云 API 往往不能直接用，本地部署几乎是唯一选择。而且如果推理量足够大，本地部署的单 Token 成本可以远低于按量付费的公有云 API，长期跑很划算。如果你用的是像 NVIDIA DGX Spark 这样的小型 GPU 工作站，在上面跑 Spark 做数据预处理、再驱动部署在本地或同网段的模型服务，整个链路都能留在内网，效率和安全性都不错。

但要注意，本地部署大模型不等于没有成本。GPU 服务器的采购或租赁费用、运维成本、模型效果和公有云大模型的差距，都要提前评估。我的经验是：先用公有云 API 把效果验证清楚，再决定要不要本地化。如果模型效果不行，省再多的推理成本也没意义。

从 Spark 代码的角度看，方案D和方案C中的“调用大模型”部分没有本质区别，都是 HTTP 调用，只是 API 地址从公网变成了内网，鉴权从 API Key 变成了内网白名单，超时和限流策略需要根据本地服务的实际情况重调。

2.5 方案对比与选型建议

把这四个方案放在一起看更清楚：

选型的时候我的思路很简单：先看数据量和频率，再看数据能不能出域，最后看团队运维能力。数据量不过百万、跑一次拉倒的，方案B足矣；数据量上千万且每天都要跑的，直接考虑方案C；数据不能出域或者成本敏感的，赶紧测方案D。方案A只适合用来验证“这条路能不能走通”。

3. 实战：PySpark 封装大模型调用的核心实现

3.1 环境准备与依赖

先交代一下实践环境。我用的是 Spark 3.4 以上版本，PySpark 作为开发语言。之所以选 Python 而不是 Scala，原因很现实：现在大模型相关的 SDK 基本都是 Python 生态，requests、openai、httpx这些库在 Python 里最好用；而且做数据清洗、prompt 拼接这类的逻辑，Python 写起来比 Scala 快得多。

你需要提前确认环境里有这些依赖：

pip install pyspark requests openai

其中openai库不是必须的，用requests也能实现，但如果你调的是 OpenAI 兼容接口，用官方 SDK 处理重试、流式这些会省不少事。我在实际项目中两种都写过，后面统一用requests做示例，因为它能把底层逻辑暴露得更清楚，方便排查问题。

还需要准备一份环境变量或配置文件，保存 API Key、接口地址、模型名。注意 Spark 跑在 YARN 集群上时，Python 进程的启动环境和提交任务的客户端环境不一定一样，API Key 最好不要直接写死在代码里。我习惯的做法是：在spark-submit的时候用--conf spark.yarn.appMasterEnv.LLM_API_KEY=xxx传进去，代码里用os.environ.get("LLM_API_KEY")读取。

spark-submit \ --master yarn \ --deploy-mode cluster \ --conf spark.yarn.appMasterEnv.LLM_API_KEY=sk-xxx \ --conf spark.yarn.appMasterEnv.LLM_BASE_URL=https://api.example.com/chat/completions \ --py-files llm_client.py \ job.py

3.2 关键代码：mapPartitions + 并发调度

下面这段代码是我在项目里精简后的通用框架，核心思路是：每个分区初始化一个客户端，分区内部用ThreadPoolExecutor控制并发，每条记录里带上业务主键，最后返回带主键的结果。

from pyspark.sql import SparkSession from pyspark.sql.types import StructType, StructField, StringType, LongType import requests import os import time import json import threading from concurrent.futures import ThreadPoolExecutor, as_completed class LLMClient: def __init__(self, api_key, base_url, model): self.session = requests.Session() self.session.headers.update({ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }) self.base_url = base_url self.model = model def chat(self, prompt, max_tokens=512, temperature=0.0): payload = { "model": self.model, "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens, "temperature": temperature, } resp = self.session.post(self.base_url, json=payload, timeout=(10, 120)) resp.raise_for_status() data = resp.json() return ( data["choices"][0]["message"]["content"], data.get("usage", {}).get("prompt_tokens", 0), data.get("usage", {}).get("completion_tokens", 0), ) def process_partition(iterator, api_key, base_url, model, max_workers=8): client = LLMClient(api_key, base_url, model) idle_thread = threading.local() def do_call(row): text_id, text = row start = time.time() result, prompt_tokens, completion_tokens = client.chat(text) latency_ms = int((time.time() - start) * 1000) return (text_id, result, prompt_tokens, completion_tokens, latency_ms) with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [executor.submit(do_call, row) for row in iterator] for future in as_completed(futures): yield future.result()

这段代码里我刻意加了一个idle_thread变量，但它其实没用到。真正要注意的是ThreadPoolExecutor的并发数和分区的数据量要匹配。如果单个分区的数据量非常大，把几万条任务一次性submit进线程池，虽然as_completed会边完成边消费，但 futures 列表本身会占内存，极端情况下也会把 Python 进程内存打爆。更稳的写法是分批提交，比如每批 200 条：

def process_partition(iterator, api_key, base_url, model, max_workers=8, batch_size=200): client = LLMClient(api_key, base_url, model) batch = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: for row in iterator: batch.append(row) if len(batch) >= batch_size: futures = [executor.submit(do_call, r) for r in batch] for f in as_completed(futures): yield f.result() batch.clear() if batch: futures = [executor.submit(do_call, r) for r in batch] for f in as_completed(futures): yield f.result()

这样既能控制并发，又能防止 futures 列表无限膨胀。

3.3 超时、重试与限流策略

大模型 API 的响应时间波动非常大。同样的模型，prompt 短的时候一两秒返回，prompt 长或者服务端排队的时候可能几十秒。所以超时设置一定要分开：连接超时短一点，10 秒就够；读取超时长一点，我一般设 120 秒，因为大模型生成一个长回答确实需要时间。

requests的timeout参数可以传元组，分别指定连接超时和读取超时：

resp = self.session.post(self.base_url, json=payload, timeout=(10, 120))

重试逻辑千万不要交给 Spark 去做，而是要在客户端层面自己实现。我建议的重试策略是：

• 429限流：必须重点试，但要退避。
• 500、502、503、504服务端异常：重点试，同样退避。
• 400、401、403、422客户端错误：不需要重试，重试多少次都会失败，直接抛出异常或者记录错误行。
• 网络异常（连接断开、DNS 解析失败）：可以重试，但要设置最大次数上限。

一个通用的指数退避重试写法：

import time import requests def call_with_retry(client, prompt, max_retries=3): for attempt in range(max_retries): try: return client.chat(prompt) except requests.exceptions.HTTPError as e: status_code = e.response.status_code if status_code in (429, 500, 502, 503, 504): sleep_time = min(2 ** attempt, 30) time.sleep(sleep_time) continue raise except requests.exceptions.ConnectionError: time.sleep(min(2 ** attempt, 30)) continue raise RuntimeError(f"调用大模型失败，已重试 {max_retries} 次，prompt 前缀: {prompt[:50]}")

注意2 ** attempt是 1、2、4、8 的指数退避，封顶 30 秒。重试次数一般 3 次就够，因为超过 3 次还失败，说明服务端可能已经挂了很长时间，硬重试只会让问题更严重。我见过有人把重试次数设成 10 次，结果服务端故障恢复后，所有积压的任务同时重试，直接把服务端打挂了。这就是典型的“好心办坏事”。

关于限流，最核心的一点是：并发数要和服务端对齐，而不是自己拍脑袋。比如模型服务支持每分钟 600 个请求，那你的 Spark 任务总并发最好控制在 10 左右（10 QPS），因为大模型单次请求的耗时通常在 2-5 秒，10 并发一分钟大约能完成 200-300 个请求，还有富余。如果开 50 并发，看起来吞吐上去了，实际上一秒内打出去 50 个请求，服务端直接 429，然后你的重试逻辑开始退避，真实吞吐反而更低。

3.4 Schema、序列化与结果对齐

调用完大模型后，数据的产出不仅仅是结果文本，最好把 Token 用量、耗时也一起带出来，后续做成本分析和性能调优都离不开这些数据。我推荐的结果 schema 至少包含这几个字段：

schema = StructType([ StructField("id", StringType(), True), StructField("result", StringType(), True), StructField("prompt_tokens", LongType(), True), StructField("completion_tokens", LongType(), True), StructField("latency_ms", LongType(), True), ])

然后在 Spark 里这样应用：

from pyspark.sql import SparkSession spark = SparkSession.builder.appName("spark_llm_inference").getOrCreate() df = spark.createDataFrame([("1", "这是一条测试文本"), ("2", "再试一条")], ["id", "text"]) result_rdd = df.rdd.mapPartitions( lambda iter: process_partition(iter, api_key, base_url, model) ) result_df = spark.createDataFrame(result_rdd, schema=schema)

这里有一个很隐蔽的坑：df.rdd.mapPartitions返回的 RDD 顺序是不保证的，尤其是多分区并行执行时，结果顺序和原始顺序完全不同。所以一定要在数据里保留业务主键，最后通过join把结果关联回原表。如果你用zipWithIndex给每一行加自增 ID，注意要先把 DataFrame 转成 RDD 再操作，或者用monotonically_increasing_id()生成唯一递增 ID，但后者只保证在同一个 DataFrame 内唯一且递增，跨执行阶段不保证连续。

如果你对输出格式要求更严格，也可以注册成 Spark SQL 的 UDF 直接用，但要接受并发的瓶颈。我的建议是：能用mapPartitions就不要用 UDF。UDF 是逐行调用的，每一行都要经过 Python 和 JVM 之间的序列化边界，性能损耗非常大。而mapPartitions是在分区级别处理，数据在 Python 进程内批量流转，序列化开销少了一个数量级。

4. 常见问题与排查实录

4.1 Spark on YARN 为什么 CPU 只能用 1 个？

这几乎是所有 Spark 新手在跑大模型任务时都会遇到的问题：提交到 YARN 上的作业，日志里显示每个 Executor 只有 1 个 vCore，任务并行度上不去，调用大模型的吞吐自然惨不忍睹。

排查思路要分清是提交参数问题还是集群配置问题。最常见的原因是spark.executor.cores没有设置，Spark 默认值是 1。在 YARN 模式下，如果你不显式指定，每个 Executor 容器就只申请 1 个 vCore。解决办法是在提交命令里加上：

--conf spark.executor.cores=4 \ --conf spark.executor.memory=8g \ --conf spark.sql.shuffle.partitions=200

另一个容易被忽略的原因是 yarn 调度器的限制。yarn.scheduler.maximum-allocation-vcores如果设置成了 1，你在 Spark 里申请再多核也没用，YARN 会直接按 1 核分配容器。这时候要改的是 YARN 的capacity-scheduler.xml或fair-scheduler.xml配置，而不是 Spark 参数。此外，如果开了 Dynamic Allocation，spark.dynamicAllocation.maxExecutors需要足够大，否则 Executor 数量也上不去。

到了 Python 场景，还有一个 Python 特有的混淆点：spark.task.cpus默认是 1，表示每个 Task 占用的 CPU 核数。如果你把这个参数设成了 2 或更高，每个 Task 会占用多个核，Executor 的并行任务数反而下降，看起来就是“明明有 4 个核，但只有一个任务在跑”。调用大模型这种 IO 密集型任务，spark.task.cpus保持 1 就好，真正的并发控制在 Python 线程池里做，这样 CPU 和网络吞吐都能打满。

4.2 Executor 任务不可序列化问题

用方案 B 写代码的时候，最容易遇到的一个异常是org.apache.spark.SparkException: Task not serializable。原因一般有两个：

第一个原因是在闭包里引用了无法序列化的对象。比如你在 Driver 上创建了一个 HTTP 客户端，然后直接在mapPartitions或者map里用它，Spark 会尝试把这个客户端序列化后发给 Executor。requests.Session里包含线程锁和连接池，根本没法序列化，所以直接报错。解决办法就是不要在闭包外创建客户端，把初始化逻辑放到每个分区内部（也就是process_partition的开头）。

第二个原因是闭包里引用了整个 SparkSession 或者一个很大的 DataFrame。有些同学会把 DataFrame 变量写在闭包里，Spark 尝试把整个 DataFrame 的逻辑计划序列化下发，轻则性能极差，重则直接报序列化错误。解决办法是把需要用的值提取成基础类型（字符串、整数、字典）再传给闭包函数。

排查这个问题有个技巧：看异常的堆栈信息，它会明确指出是哪个对象无法序列化。如果堆栈里提到了LLMClient，那基本就是客户端初始化位置不对；如果提到了SparkSession或者DataFrame，那闭包里多半不小心引用了不该引用的对象。

4.3 429 限流与 5xx 服务端错误

调大模型 API 时，429 Too Many Requests是高频错误，尤其是跑批任务时。这个状态码表示你请求太频繁，服务端已经限制你了。很多人的第一反应是加大重试次数，其实这完全搞反了。429 的正确处理方式是降低并发，不是增加重试。

我踩过一次很经典的坑。某个任务配了 16 并发，跑了几分钟就大量 429，重试逻辑开始指数退避，退避之后又叠加新请求，服务端的限流计数器一直被顶在最大值，结果整个任务跑了三个小时还没跑完，Token 消耗倒是不少。后来把并发降到 4，同样数据量四十分钟跑完了。这个故事的核心教训是：大模型服务的并发能力是有限的，你只能让 Spark 去适配它，而不是让服务端来适配你的 Spark 并行度。

遇到 5xx 错误时，情况不太一样。500、502、503、504 通常意味着服务端正在经历故障或过载。这时候可以重试，但要给服务端留出恢复时间。我建议 503 和 504 的退避时间至少 5 秒起步，最多 60 秒；如果连续 3 次重试仍然失败，就把这批数据标记为失败，写入错误表，而不是无限重试卡住整个作业。

4.4 Executor OOM 与内存模型调优

Spark 调用大模型时的 OOM，有一大半跟模型输出结果太大有关。大模型单次返回可能包含几千甚至上万 Token 的文本，如果每条记录的 prompt 本身就长，结果也长，Python 进程里有成百上千条这样的数据排队等待处理，内存很容易被打爆。

要理解这个问题的本质，得稍微提一下 Spark 的内存模型。JVM 堆内存被划分为 Reserved、User Memory、Spark Memory（又分 Execution Memory 和 Storage Memory）等区域。Spark SQL 的 DataFrame 操作主要使用 Execution Memory，而 Python UDF 的进程内存并不完全受 Spark 的堆内存管控，它由spark.executor.pyspark.memory参数单独限制。如果你跑的是 PySpark 调用大模型，可能 OOM 的其实是 Python 进程，而不是 JVM。

配置上可以做两件事：一是适当调大spark.executor.pyspark.memory，默认值只有 512MB，对大模型任务来说偏小；二是控制单个分区的数据量，避免一次性把整个分区的数据加载到内存。我在实践中通常把mapPartitions内部按 200 条一批处理，每批处理完就释放，内存占用稳定很多。

另外，输出结果尽量不要做“大字段的频繁转换”。比如把几万 Token 的长文本反复做 JSON 解析、正则替换、子串截取，Python 进程的内存会很快膨胀。一次性加工完再返回，比拆成多步操作更省内存。

4.5 其他容易被问到的点

最后说一个面试角度。Spark 怎么调用大模型这个问题，面试官真正想听的其实不是requests.post怎么写在 UDF 里，而是你能不能答出“为什么 UDF 直连不好”“并发怎么控制”“失败怎么处理”“结果怎么对齐”这几个关键点。

我建议的答题结构是：先说场景分类，明确“这是离线批量推理”，再说方案选型，强调mapPartitions加客户端池化是工程上的主流选择，最后补充并发、重试、成本统计这些落地的细节。如果面试官追问“为什么不能直接用 UDF”，你就把连接复用、序列化、并发控制这三点讲清楚，基本就能让人信服。如果面试官继续问“数据量上亿怎么办”，你可以接上消息队列解耦和本地推理服务那套，说明你确实从系统层面想过这个问题。这个问题现在几乎成了数据开发岗位的高频题，准备好吃透固然重要，但更关键的是你要有真实项目里的数据来判断方案的边界在哪里。

5. 工程化落地：监控、成本与稳定性治理

5.1 可观测性：日志、累加器与调用明细表

代码能跑通只是第一步。真正放到生产环境里，你还需要回答几个问题：任务跑到哪了？还有多少条没跑？失败了多少条？花了多少 Token？如果要回答这些问题，必须在设计的时候就埋好观测点。

最简单的做法是用 Spark 的Accumulator来做进度统计。比如定义三个累加器，分别统计成功条数、失败条数、重试次数：

from pyspark.sql import SparkSession spark = SparkSession.builder.appName("spark_llm_inference").getOrCreate() success_counter = spark.sparkContext.accumulator(0) fail_counter = spark.sparkContext.accumulator(0) retry_counter = spark.sparkContext.accumulator(0)

然后在调用逻辑里根据结果更新累加器，作业结束后通过success_counter.value就能看到最终的调用统计。这个方法比较简单，但它的局限性也很明显：累加器的值只能在作业结束后拿到，任务跑的过程中你是看不到实时进度的。如果想做实时监控，我建议把每次调用的明细写回一张结果日志表，字段包括id、prompt长度、result长度、prompt_tokens、completion_tokens、latency_ms、status、error_msg、retry_count。这样既能看到单次调用的情况，也能后续通过 SQL 做聚合分析。

日志方面给一个实用建议：不要在生产日志里打印完整的 prompt 和结果。一方面是因为数据敏感，另一方面是日志文件会快速膨胀。我一般只打印id前缀、耗时、Token 数、状态码这些元信息。如果有特殊需要排查某一条数据，可以通过id去查明细表。

5.2 成本控制：缓存、降级与模型分级

调用大模型的成本是真实存在的，尤其是公有云 API 按 Token 计费，跑一次千万级数据的任务，账单可能是几万甚至几十万。所以成本控制不是后端优化的问题，而是方案设计阶段就要考虑的。

第一个手段是输入缓存。很多真实业务场景里，相似的文本会被重复提交。比如同一个文案在多个表里出现，或者同一个用户的多条记录用的是同一段模板。可以在 Spark 作业开头先按文本内容做一次去重，只对去重后的文本调用大模型，再通过关联把结果映射回全量数据。如果业务场景里重复率高，这个优化能把成本直接砍掉一半以上。

第二个手段是结果缓存和增量处理。如果这个任务是每天跑的，那不应该每次把全量数据都跑一遍。可以设计一个结果表，记录每个业务 ID 最近一次处理的时间，下次调度只处理新增和更新的数据。这样既节省费用，也缩短了作业时间。

第三个手段是模型分级。不是所有请求都需要最强的模型。简单的情感分类用 7B 模型就够了，复杂的长文档摘要才需要 70B 甚至更大的模型。可以在数据准备阶段加一个路由逻辑，按任务难度、输入长度、业务重要性把数据分成几档，各调度到不同模型上。这个思路在业务量特别大的场景下很有价值，也是我比较推荐的一种“架构级成本优化”。

5.3 模型部署与迭代：本地模型还是 API？

最后聊一下模型服务的形态。公有云 API 的好处是省心，效果通常也比较好，但问题是要考虑数据合规，而且随着调用量增大，成本是线性增长的。本地部署模型的好处是数据不出域、单 Token 成本低，但运维负担重，你既要做显存规划，又要处理并发队列、负载均衡、模型版本的滚动升级。

我的建议是：先用公有云 API 跑通业务，验证产品价值，等技术方案稳定后再考虑本地部署。如果团队里有人熟悉 GPU 集群运维，或者数据合规要求非常严格，就提前做本地化的选型测试。现在用 VLLM 或者 TGI 部署开源模型已经非常简单，接口也是 OpenAI 兼容的，Spark 这边的代码几乎不用改，只需要换一下base_url和model就行。这一点是这个方案最舒服的地方。

如果你用的是 Spark 3.4 以上的版本，并且 GPU 资源就在本地集群，也可以尝试把 GPU 调度接入 Spark，让 Executor 直接申请 GPU 资源跑推理。但对大多数团队来说，把 GPU 推理做成独立的服务，Spark 只作为客户端去调用，是更简单、也更符合“大模型服务化”这个趋势的做法。后面模型要升级、要灰度、要回滚，都在推理服务这一侧独立完成，Spark 作业本身不会有感知。

6. 最后一些个人体会

文章写到这，技术上的东西基本讲完了，最后分享一点我自己的方法论。做这类项目，我最大的体会是永远先跑一条链路再放大。不管方案设计得多完美，第一次跑任务时一定先取几百条数据、用很小的分区数验证整个链路能不能通，再逐步放大。我有一个同事就是跳过了这一步，直接拿全量千万级数据跑，结果 API Key 配错了，作业跑了两小时才发现全是认证失败，浪费了大量时间。

另外一个体会是，大模型调用的代码本身没那么难，难的是把它嵌入到已有的数据链路里还不破坏稳定性。你要面对的是 Spark 的资源调度、Python 进程的内存边界、外部服务的限流和抖动、数据质量的参差不齐。这些问题的排查往往比写代码更花时间。所以我建议在做架构设计时，一定要把监控、重试、限流、成本统计当成一等公民，而不是最后补丁式地加上去。

如果你正在做类似的项目，希望这篇文章能帮你少踩几个坑。有什么更好的方案或者我没覆盖到的场景，欢迎在评论区交流，我这几年积累了不少稀奇古怪的报错样本，说不定还能再写一篇问题排查合集。