EMR Serverless Spark 上线 GPU 规格之后，很多团队的认知还停留在“Spark 是纯 CPU 计算引擎”这个阶段。其实在 Spark 的 RDD 和 DataFrame 分布式框架里，把 GPU 作为可调度的 executor 资源，已经是很成熟的做法了。今天这篇东西，我结合自己跑的 CPU + GPU 异构计算任务，把 EMR Serverless Spark 上的配置、镜像、调度和调优完整走一遍。后面所有内容都是以可复现为目标，看完你可以直接照着自己集群的版本改参数。
先说清楚它能解决什么问题：以前图像特征提取、向量化推理、小规模微调这类重计算任务，要么单独起 GPU 集群，要么在 Spark 里用 CPU 算到天荒地老。现在 EMR Serverless Spark 支持在同一个作业里同时申请 CPU 和 GPU，让 Spark 做数据编排，GPU 做张量计算，两条腿走路。适合做数据管道里需要 GPU 加速的团队，也适合从自建 Hadoop 迁到 Serverless 平台的开发者。
1. 为什么在 Serverless Spark 里引入 GPU：CPU 和 GPU 的角色分工
1.1 传统 Spark 任务为什么吃不满 CPU
以前调优 Spark 任务，大家关注的是 executor 数量、core 数、内存、shuffle 分区。这套模型默认计算单元是 CPU，跑 TPC-DS、ETL、普通聚合分析没问题。但你一旦把模型推理、图像缩放、向量化算子放进来，CPU 就力不从心了。
典型场景是：一张 Redis 里读出来的用户特征表，要跑一个几百 MB 的深度学习模型做打分。Spark 每个 task 处理一个 partition，每条记录调用一次模型推理。单个 CPU core 推理一次大概几十毫秒，看起来不慢，可一百万个 partition 意味着几百万次推理，累积起来就是几小时。即使做 batch inference，CPU 的浮点吞吐和 GPU 差一到两个数量级。瓶颈不在 Spark 调度，而在计算硬件。
加 CPU 核数能解决一部分，但不是最优解。CPU 核越加越多，每个核还要配内存，Spark 内存开销线性上涨，资源单价也上去了。相反，用 GPU 做矩阵运算，显存占用稳定，吞吐高得多。异构计算进入 Spark 的初衷，不是去替代 Spark 的分布式调度，而是把最重的数值计算从 CPU 上卸下来。
1.2 GPU 适合放进 Spark 工作流的什么位置
异构计算里最常见的做法是“CPU 管 I/O 和编排，GPU 管计算”。在 Spark 作业里，数据读取、解析、过滤、shuffle、聚合这些操作仍然由 CPU 执行；真正需要大规模并行数值计算的部分，比如：
- 图像和视频的预处理，特别是 resize、归一化、复杂增强；
- 大模型的 embedding 提取和 batch 推理；
- PyTorch / TensorFlow 的训练循环；
- 向量相似度计算和特征工程里的高维矩阵运算。
这些计算放到 GPU 上，Spark 只是负责把数据分发到 executor，并在每个 executor 内协调 GPU 调用。GPU 不直接替换 Spark 的 task 调度器，而是作为 executor 上的一个资源卡，被 Spark 的资源调度框架感知、分配和隔离。这样你的作业从外部看仍是 Spark 作业，调度、监控、日志体系全部复用，不用另起炉灶。
1.3 EMR Serverless Spark 对异构支持的定位
EMR Serverless Spark 的做法比较省心：平台管理 Spark 集群的生命周期，你只需要提交作业。它支持把 GPU 作为一种可申请的资源规格，作业指定需要的 GPU 数量后，executor 启动时就会挂载对应的 GPU 设备，同时自动处理 NVIDIA 驱动、容器运行时等底层细节。
我自己体验下来，它和自建集群最大的区别在两点。第一，你不用预先规划 GPU 机器数量，提交作业时按需申请，作业结束资源立刻释放，特别适合一天只有几个时段的推理管道。第二，平台自带资源隔离，多个作业跑在同一个基础设施上，不会因为别的团队抢 GPU 导致你显存爆掉。代价是底层的 Spark 配置项不完全暴露，某些自定义资源脚本需要你通过镜像或 Spark 参数注入，这部分后面会细说。
2. 第一台异构作业怎么跑起来：配置与提交
2.1 控制台提交中的 CPU + GPU 资源形态
在 EMR Serverless Spark 控制台创建作业时，计算资源配置通常分两块：executor 数量、单个 executor 的 CPU 和内存。启用 GPU 后，还会多出一项 GPU 数量。很多网络热词里提到的“gpu租用”“服务器gpu推荐”就是这种按资源形态计费的模式。
我的建议是：不要一上来就申请整机 8 卡。先从单 executor 单 GPU 开始，也就是把 GPU 数量设置为 1，CPU 设置到 4~8，内存给 16~32 GB。这样每个 executor 独占一张 GPU，代码逻辑里直接通过 CUDA 可见设备编号拿到当前进程对应哪张卡，简单可靠。等你确认任务能吃满 GPU 了，再去考虑一个 executor 多卡。
在作业配置里，还有两个和 Spark 资源相关的开关需要注意：
- 是否启用动态分配：建议关闭。带 GPU 的作业如果开动态分配，executor 会随时增减，GPU 资源规划会变得很难做，而且缩容时 GPU 显存里的模型要重新加载，反而慢。
- spark.executor.cores 不要设置得比物理核数还高。常见配置是 executor 给 4 CPU，跑两层并发，内层还能开线程做数据预处理，生产环境实测比较稳。
2.2 CLI 与 SDK 提交的关键参数
如果不想用控制台，更建议用 CLI 或 SDK 做参数化提交。控制台每次都点一堆配置，麻烦且不好纳入 CI/CD。CLI 提交时，关键是构造 Spark 参数里的资源相关配置。
标准 Spark 里 GPU 资源需要三个东西共同作用：spark.executor.resource.gpu.amount、spark.executor.resource.gpu.discoveryScript，以及 driver 对应的资源设置。EMR Serverless 的托管环境通常预置了 GPU 发现脚本，但为了保险，我习惯在 Spark 参数里显式覆盖：
aws emr-serverless start-job-run \ --application-id "${APP_ID}" \ --execution-role-arn "${ROLE_ARN}" \ --job-driver '{ "sparkSubmit": { "entryPoint": "s3://bucket/scripts/gpu_check.py", "sparkSubmitParameters": "--conf spark.executor.cores=4 --conf spark.executor.memory=16g --conf spark.executor.instances=2 --conf spark.executor.resource.gpu.amount=1 --conf spark.executor.resource.gpu.discoveryScript=/opt/spark/scripts/gpu_discovery.sh --conf spark.task.resource.gpu.amount=1" } }' \ --configuration-overrides '{ "monitoringConfiguration": { "s3MonitoringConfiguration": { "logUri": "s3://bucket/logs/" } } }'这个命令里的 spark.task.resource.gpu.amount 很多人会漏掉。它的作用是告诉 Spark，每个 task 启动时要求拿到 1 个 GPU 地址。没有这个参数，executor 虽然申请了 GPU，但 task 调度时不会感知 GPU，你代码里直接调 nvidia-smi 也许能看到卡，但 Spark 层面并没有真正把它当作可调度资源来排队和隔离，多任务并发时会互相抢占显存。
2.3 确认 GPU 真的被 Spark Executor 看到
配置完了第一件事不是跑业务，而是验证 GPU 可见性。写一个超简单的 PySpark 脚本，把每个 executor 上识别到的 GPU 信息打印出来：
from pyspark.sql import SparkSession import os, subprocess spark = SparkSession.builder.appName("gpu_check").getOrCreate() def check_gpu(_): gpu_ids = os.environ.get("CUDA_VISIBLE_DEVICES", "UNSET") nvsmi = subprocess.check_output(["nvidia-smi", "-L"]).decode("utf-8") return [(gpu_ids, nvsmi)] df = spark.sparkContext.parallelize(range(4), 4).map(check_gpu).toDF(["cuda_visible_devices", "nvsmi"]) df.show(truncate=False)跑完看你日志里的输出：每个 partition 对应一个 executor 时，CUDA_VISIBLE_DEVICES 应该分别是 0 或对应 GPU 编号，nvidia-smi 能看到卡型号。如果 CUDA_VISIBLE_DEVICES 是空或者显示 UNSET，说明 GPU 资源没注入到容器，先查镜像和资源申请配置，不用继续往下做。
3. 镜像与运行时：把 CUDA/PyTorch 环境装明白
3.1 基础镜像选择与 CUDA 版本
EMR Serverless Spark 官方提供了基础镜像，里面预装了 Spark 运行环境和 Hadoop 依赖，但不一定带 CUDA 工具链。你在镜像里跑 GPU 计算，必须自己装 CUDA、cuDNN，以及按需的 PyTorch 或 TensorFlow。
选 CUDA 版本的时候，不要直接装最新版。先看你要跑的深度学习框架支持哪个 CUDA 版本。比如你打算用 PyTorch，官方安装命令会明确标注支持 CUDA 11.8 还是 12.1，你就照着那个版本装。镜像里的 NVIDIA 驱动由底层宿主机提供，容器内只需要 CUDA runtime 和 toolkit，不需要装驱动，装了反而可能和宿主机驱动版本冲突。
我在 EMR Serverless 上用过的比较稳的组合是：基础镜像选择 EMR 7.2.0 对应的 Spark 镜像，Python 3.9 或更高，CUDA 跑 11.8 或 12.1，PyTorch 用对应的 cu118 或 cu121 版本。这套组合在多个任务里都跑通了，没有出过底层 ABI 不兼容的问题。
3.2 写一个带 GPU 验证的 Dockerfile
镜像构建过程本身不复杂，但有几个细节很关键。下面这个 Dockerfile 是可用的参考模板：
FROM public.ecr.aws/emr-serverless/spark/emr-7.2.0:latest USER root # 安装基础工具 RUN yum install -y wget tar gzip && yum clean all # 安装 CUDA runtime（不安装驱动） RUN wget https://developer.download.nvidia.com/compute/cuda/repos/rhel8/x86_64/cuda-toolkit-12-1-12.1.1-1.x86_64.rpm \ && yum install -y ./cuda-toolkit-12-1-12.1.1-1.x86_64.rpm \ && rm -f cuda-toolkit-12-1-12.1.1-1.x86_64.rpm ENV PATH="/usr/local/cuda/bin:${PATH}" \ LD_LIBRARY_PATH="/usr/local/cuda/lib64:${LD_LIBRARY_PATH}" # 安装 PyTorch RUN pip3 install --no-cache-dir torch==2.1.0 --index-url https://download.pytorch.org/whl/cu121 # 创建 Spark 用户需要的目录，避免运行时权限问题 RUN mkdir -p /opt/spark/scripts && chmod 755 /opt/spark/scripts # GPU 发现脚本：Spark 通过该脚本识别每个 executor 上的 GPU 设备 RUN printf '#!/bin/bash\nnvidia-smi --query-gpu=index --format=csv,noheader | nl -v 0 | awk '\''{print "{\"name\": \"gpu\", \"address\": \"" $2 "\", \"index\": " $1-1 "}"}'\''\n' > /opt/spark/scripts/gpu_discovery.sh \ && chmod +x /opt/spark/scripts/gpu_discovery.sh USER hadoop最后这个 GPU 发现脚本是参考了 Apache Spark 官方 GPU 发现脚本的写法。它的任务是输出一个 JSON 数组，Spark 会拿这个内容建立 GPU 地址列表，然后分配给不同 executor 和 task。不同平台的脚本路径会有差异，如果你不确定，可以先用一个简单的nvidia-smi -L输出做调试，但生产环境建议还是按照 Spark 要求的 JSON 格式来写，否则资源隔离规则不生效。
3.3 镜像里容易踩的坑
镜像构建最大的坑是用户权限。基础镜像默认是非 root 用户 hadoop，很多 pip 和 yum 操作需要 root，如果你直接在 Dockerfile 里切到 hadoop 再装东西，会碰到权限报错。正确顺序是切 root、安装、改权限、最后切回 hadoop。
第二个坑是 CUDA 版本和驱动版本不匹配。容器里 CUDA runtime 要求宿主机驱动版本不低于某个下限，EMR Serverless 的宿主机驱动通常比较新，但为了保险，不要用太老的 CUDA 版本。比如 CUDA 10.x 在老卡上可能没问题，但新平台的驱动不一定向前兼容到那个版本。
第三个坑是镜像体积。EMR Serverless 拉取镜像占启动时间，PyTorch 加 CUDA 动辄几个 GB，如果基础镜像在海外仓库，第一次作业启动会慢得让人怀疑人生。建议把镜像推到离你 region 近的镜像仓库，作业运行时指定私有仓地址，减少拉取时间。
4. 在 Spark 上完成 GPU 计算：三个典型场景拆解
4.1 场景一：用 GPU 跑 ETL 里的图像特征提取
假设你有一个 S3 里的 image metadata 表，里面有图片路径、业务标签等字段。你要给每张图片生成一个 embedding 向量，然后把向量写回表里做后续检索。在纯 CPU 环境下，这个任务用 OpenCV 加 ONNX Runtime，可能要跑几小时；改成 GPU 后，在 Spark executor 内分配好模型，每个 task 批量处理图片，性能提升非常明显。
代码思路是这样：先用 Spark 读取图片路径列表，repartition 保证每个 task 处理的数据量可控，然后在 mapPartitions 里加载一次模型，对整个 partition 的图片做 batch inference，最后把 embedding 向量收集成 DataFrame 写回。关键点是模型加载次数一定要少，不能每条 record 加载一次。
import pandas as pd import torch from pyspark.sql import SparkSession, Row from pyspark.sql.types import StringType, ArrayType, FloatType spark = SparkSession.builder.appName("gpu_feature_extract").getOrCreate() def extract_embeddings(iterator): import torch import torchvision.transforms as T from PIL import Image import io device = "cuda" if torch.cuda.is_available() else "cpu" model = torch.load("s3://bucket/models/resnet18.pt", map_location=device) model.eval() transform = T.Compose([ T.Resize((224, 224)), T.ToTensor(), T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) ]) for row in iterator: image_bytes = row.image_bytes img = Image.open(io.BytesIO(image_bytes)).convert("RGB") tensor = transform(img).unsqueeze(0).to(device) with torch.no_grad(): emb = model(tensor).cpu().numpy().tolist()[0] yield Row(image_id=row.image_id, embedding=emb) input_df = spark.read.format("parquet").load("s3://bucket/input/images") output_df = input_df.rdd.mapPartitions(extract_embeddings).toDF(["image_id", "embedding"]) output_df.write.mode("overwrite").parquet("s3://bucket/output/embeddings")这里有几个经验：图片字节不要在 DataFrame 里反复序列化，最好用二进制字段一次性读入；batch inference 还有优化空间，但 mapPartitions 里写成一个循环，配合 PyTorch 的 no_grad，已经能跑满 GPU 利用率。如果你发现 GPU 利用率一直不高，多半是单次推理的数据量太小，可以把多个图片拼成 batch，或者调大 spark.task.resource.gpu.amount 对应的分区粒度。
4.2 场景二：在 Executor 内启动 PyTorch 训练
除了推理，EMR Serverless Spark 也可以用来做小规模训练。这里说的不是用 Spark 框架重写训练逻辑，而是把 Spark 当作分布式任务编排层：每个 executor 内启动一个 PyTorch 训练进程，训练数据从 Spark DataFrame 分发。
比较适合这种模式的场景是：多个业务线各自要微调一个小模型，每个模型数据量不大，但模型数量很多。你写一个 PySpark 作业，外层遍历模型训练参数，内层通过 mapPartitions 把训练数据发给每个 executor 的 PyTorch trainer，训练完成后把指标写回结果表。这样一个作业可以批量训练几十个模型，GPU 利用率比单独起几十个 Pod 高得多。
训练代码的写法：
def train_model(partition_data): import torch import torch.nn as nn device = "cuda" if torch.cuda.is_available() else "cpu" # 每个 partition 只初始化一个模型 model = SimpleModel().to(device) optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) loss_fn = nn.MSELoss() for batch in partition_data: x = torch.tensor(batch["features"], dtype=torch.float32).to(device) y = torch.tensor(batch["labels"], dtype=torch.float32).to(device) optimizer.zero_grad() output = model(x) loss = loss_fn(output, y) loss.backward() optimizer.step() yield {"loss": loss.item()}注意：executor 内训练时，数据 shuffle 和 batch 切分逻辑要自己处理，Spark 负责的是分区层面的并行。如果一个分区里的训练数据太大，GPU 显存放不下，建议先按固定行数做 groupBy 或 repartition，把数据切小再映射到 partition 上。
还有一个容易犯的错误：在 executor 里使用 PyTorch DataLoader 同时又开多进程。GPU executor 的 worker 数量如果超过 GPU 显存承载能力，会直接 OOM。这里更简单的做法是让每个 executor 进程对应一个 GPU，内部不开多进程 DataLoader，全部用单进程加载，数据集不大时性能足够。
4.3 场景三：Spark SQL + GPU UDF
有些场景不想写 mapPartitions，希望直接在 SQL 里调用 GPU 函数。Spark 的 UDF 机制天然支持：注册一个 Python UDF，内部调用 GPU 算子。但需要注意，UDF 的调用粒度是每行一个函数调用，如果你在 UDF 里反复加载模型，性能会崩塌。
推荐做法是“UDF 里不做模型加载，只做张量计算”。你可以把模型做成全局单例，首次加载后缓存下来，后续 UDF 调用直接使用。Spark 3.x 里可以用 pandas UDF，批量传入多行，显著降低 Python 函数调用开销。
from pyspark.sql.functions import pandas_udf import pandas as pd @pandas_udf("float") def gpu_predict_udf(features: pd.Series) -> pd.Series: global _model if _model is None: _model = load_gpu_model() # 将 pandas Series 转成 tensor，做 batch 推理 return predict_on_gpu(_model, features) spark.udf.register("gpu_predict", gpu_predict_udf) spark.sql("SELECT id, gpu_predict(features) AS score FROM inference_table")petastorm 或者 spark-rapids 这类库也能做 GPU SQL 加速，但那是另一个方向。我这里讲的是你自己写 UDF 控制计算逻辑，灵活性更高，适合不太规则的模型推理。
5. 性能调优、资源估算与问题排查
5.1 显存和计算资源怎么估算
申请 GPU 之前，先算清楚单任务需要多少显存。以 PyTorch 推理为例，可以粗略估：模型参数量乘以 2~4 字节，得到一个权重占用；batch size 乘以单条样本的激活量，得到临时显存占用；两个加起来，再留 20% 余量。
举一个实际例子：ResNet50 参数量约 25M，FP32 权重约 100 MB。输入 batch size 64、224x224 的三通道图，激活显存大概在 300 MB 左右。那么单卡 16 GB 的 GPU 跑这个任务绰绰有余。如果是 LLaMA 类大模型，7B 参数 FP16 权重就要 14 GB，单卡 16 GB 很紧张，推理时还要算 KV cache，建议直接考虑多卡分片或者用量化模型。
EMR Serverless Spark 里，GPU 数量和 executor 数量是两维。假设你有 4 个 executor，每个 executor 申请 1 卡，那么同一时间最多有 4 个 task 在跑 GPU 计算。每个 task 内部再决定 batch size。你不要天真地以为申请了 4 卡就能同时跑 100 个并发推理，并发度还是受 Spark task 数量限制。
5.2 作业失败和性能毛刺的排查思路
在 Serverless 环境里排查 GPU 问题，比自建集群麻烦在没有宿主机访问权限，所以日志要尽量提前做好。常见排查路径有三条：
第一条是看 Spark executor 日志。如果容器启动阶段失败，通常是镜像问题，重点看 stderr 里有没有 CUDA driver 相关报错。比如CUDA initialization: CUDA unknown error，大概率是镜像内 CUDA runtime 和宿主机驱动不兼容。
第二条是看 nvidia-smi 输出。建议在每个 task 开始和结束时，把显存占用、GPU 利用率写到自定义 metrics 里。EMR Serverless 支持自定义 metrics，你可以用 Prometheus 格式推送，然后再做可视化。没有监控数据的异构作业，出了问题基本只能靠猜。
第三条是看 Spark UI 的 task 耗时分布。如果 GPU task 的耗时方差很大，大概率是数据倾斜，某个 partition 的数据量明显大于其他 partition，导致 GPU 在那里空转。这时候要对数据做 repartition 或按 bucket 重新划分，不要让一个 task 处理别人三倍的数据。
5.3 常见问题速查表
| 现象 | 可能原因 | 解决方法 |
|---|---|---|
| nvidia-smi 能看到卡，但 torch.cuda.is_available() 为 False | PyTorch 和 CUDA 版本不匹配 | 用匹配的 cu118/cu121 版本重装 torch |
| 作业启动特别慢 | 镜像体积大或镜像拉取慢 | 缩小镜像、推送到 region 内私有仓库 |
| 多个 task 之间显存冲突 | 没有配置 spark.task.resource.gpu.amount | 显式配置 task 资源，用 GPU 地址发现脚本 |
| GPU 利用率长期低于 30% | task 数据量太小，算子没吃饱 | 增大 batch size、合并小 partition、减少模型重复加载 |
| 报错找不到 libcuda.so.1 | 镜像里缺 CUDA runtime 或 LD_LIBRARY_PATH 不对 | 检查 LD_LIBRARY_PATH，确认 libcuda.so 存在 |
6. CPU 与 GPU 成本与选型建议
6.1 什么时候继续用 CPU
CPU + GPU 异构不是银弹。对于纯 SQL ETL、大表 Join、复杂 Shuffle 这类任务，GPU 发挥不出优势，反而因为 GPU 规格单价高，成本会明显上升。我见过有人把简单的 groupBy 也跑到 GPU 上，结果任务变慢了，原因是 GPU 资源申请、内核启动和数据传输的开销远远大于计算本身带来的收益。
如果你发现任务的计算特性是“高 I/O、低算力”，比如大量网络请求、S3 文件扫描、正则匹配，那老老实实用 CPU executor。异构计算的价值在于用计算换时间，如果计算占比本身就很低，换了也白换。先用 Spark UI 估算一下 CPU time 和 task 时间的比例，如果 CPU time 占比不到 50%，GPU 大概率帮不上忙。
6.2 异构任务成本评估思路
EMR Serverless 按资源小时计费，CPU 和 GPU 是分开计价的。评估一个异构任务是否划算，不能只看作业跑得快了多少，还要看单位时间成本的变化。假设 CPU 任务跑 10 小时，每小时成本 5 元，总成本 50 元；GPU 任务跑 2 小时，每小时成本 15 元，总成本 30 元，这才叫划算。
另外要算上人工成本：GPU 任务的镜像维护、依赖升级、异常排查，通常比纯 CPU 任务复杂。如果团队没有容器和 CUDA 基础，第一周可能都在踩环境坑。建议先挑一个最耗时、最稳定的单模型推理任务做试点，跑通后再扩大到其他场景，不要一上来就把整个数据管道全部切到 GPU。
调参方面，单 executor 的 GPU 数量和 CPU 数量的比例，不要照抄别人的方案。我用过的比较合理的配置是：CPU 给 8 核、内存 32 GB、GPU 给 1 张卡。这种配置下，CPU 有足够余量做数据读取和预处理，GPU 专注于矩阵运算，两者基本不会互相拖累。如果 CPU 太少，会出现 GPU 在等数据的局面，Performance 反而比不上纯 CPU。
7. 一点实践体会
最后说个我每次跑异构作业都会检查的点：task 数量和 GPU 数量的匹配。很多人喜欢把数据切成几千个 partition，认为并行度越高越快。但在 GPU 场景下，每个 task 启动和销毁都有开销，而且 task 太多会导致每个 task 处理的数据太少，GPU 流水线根本没建立起来。我现在的习惯是让 partition 数量等于 executor 总数乘以每 executor 并发数，必要时把单 task 处理的数据量放在“能跑 3~5 秒”这个量级，GPU 利用率明显比乱切 partition 高很多。
还有镜像缓存，EMR Serverless 对私有镜像的拉取策略会影响作业启动时间。我测试了几次之后，会把常用依赖打进镜像，而不是每次启动时 pip install。比如 torch、torchvision、opencv 这种固定版本，装好后基本不动；业务代码则通过 Spark entry point 从 S3 拉取，这样既能保证环境稳定，又能让业务代码迭代不需要重新构建镜像。这套模式实践下来，作业冷启动时间至少能压掉三分之一。
CPU + GPU 异构计算在 Serverless Spark 里已经不是新鲜概念，但真正用好的人不多。关键在于理解 Spark 资源调度和 GPU 资源调度的边界，以及在实际运行中不断根据监控数据调整任务切分和 batch size。在这套架构下跑熟之后，你会发现很多原本要单独搭 GPU 集群才能解决的批处理问题，现在一个 Spark 作业就能搞定。
