从本地 Hadoop 到托管 Spark：PayPal 如何重构大规模分析平台

当数据平台每天处理 PB 级数据时，“还能运行”并不等于“足够敏捷”。PayPal 原有的本地分析体系能够支撑庞大业务，但随着业务增长、收购整合和技术栈叠加，集群扩容、作业维护与跨平台数据流转逐渐拖慢了洞察速度。

PayPal 的应对方式，是将分析负载从本地 Hadoop 平台迁移到 Google Cloud 的 Managed Service for Apache Spark，并围绕 Spark、Cloud Storage 和 BigQuery 建立更统一的云原生数据基础。结果不仅是资源供给更快：核心分析负载的处理时间改善了 25%，SLA 达标率提升了 30%，工程团队也减少了基础设施维护和故障救火。

真正的瓶颈不是算力，而是平台碎片化

PayPal 的旧平台并不弱。它每天可以处理 PB 级数据，问题在于多年演进形成了不均匀的技术版图：不同平台、集成方案和批处理流程分别解决过具体问题，但叠加之后带来了三类成本。

• 扩容周期过长：面对零售旺季或全球产品发布，硬件采购与人工配置可能需要数月。
• 资源利用率难平衡：为了应对峰值而长期保留容量，会产生闲置成本；容量不足又会威胁 SLA。
• 工程复杂度持续上升：同类任务运行在不同平台上，团队需要维护多套部署、监控和故障处理方式。

这种环境很容易形成“复杂性导致停滞”的循环：平台越复杂，任何调整需要协调的系统越多；变更越慢，团队越倾向于继续沿用旧流程。

因此，这次现代化不能只理解为把 Hadoop 作业复制到云上。更重要的变化是统一执行引擎、资源供给方式、数据接口和运维模型。

托管 Spark 改变了哪些关键约束

1. 从预置容量转向按作业供给资源

托管 Spark 可以在数分钟内启动计算环境，并根据作业需求弹性扩展。团队不再需要为了偶发峰值长期运行大规模集群，也不必等待人工安装和配置硬件。

这对周期性业务尤其重要，例如季末结算、促销活动、风险模型重算或全球发布。资源可以围绕负载生命周期存在，而不是围绕服务器生命周期存在。

2. 用统一接口收敛不同工作流

标准化 Apache Spark，使不同团队能够复用相同的开发模型、作业提交方式和性能调优知识。配合托管服务后，补丁、基础设施配置和部分运行时维护不再由每个业务团队重复承担。

统一并不意味着所有任务都必须写成同一种代码，而是让批处理、聚合和数据转换共享一套可治理的执行底座。

3. 缩短数据移动链路

Managed Spark 与 Cloud Storage、BigQuery 等 Google Cloud 服务的原生集成，可以减少中间导出、文件搬运和自建连接器。数据链路越短，失败点、重复存储和数据新鲜度延迟通常也越少。

这种统一数据基础同样有助于后续的智能体应用。Agentic 系统需要可发现、及时且受治理的数据；如果底层仍由大量孤岛和手工同步任务组成，再强的模型也难以稳定获取可信上下文。

可以这样实践：提交一个无常驻集群的 PySpark 批任务

下面给出一个可改造的最小示例。假设 Cloud Storage 中已有交易 Parquet 文件，字段包含 transaction_id、event_time、amount 和 status。脚本按小时聚合成功交易，并把结果写回 Cloud Storage。

将以下内容保存为 hourly_transactions.py：

import argparse

from pyspark.sql import SparkSession
from pyspark.sql import functions as F


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True)
    parser.add_argument("--output", required=True)
    return parser.parse_args()


def main():
    args = parse_args()
    spark = SparkSession.builder.appName("hourly-transactions").getOrCreate()

    transactions = spark.read.parquet(args.input)

    hourly = (
        transactions
        .filter(F.col("status") == "COMPLETED")
        .withColumn("event_hour", F.date_trunc("hour", F.to_timestamp("event_time")))
        .groupBy("event_hour")
        .agg(
            F.countDistinct("transaction_id").alias("transaction_count"),
            F.sum("amount").alias("total_amount"),
        )
        .orderBy("event_hour")
    )

    (
        hourly.write
        .mode("overwrite")
        .partitionBy("event_hour")
        .parquet(args.output)
    )

    spark.stop()


if __name__ == "__main__":
    main()

运行前需要准备一个 Google Cloud 项目、一个 Cloud Storage 存储桶，以及具备 Dataproc 和存储访问权限的服务账号。把下面变量替换成自己的值：

export PROJECT_ID="your-project-id"
export REGION="us-central1"
export BUCKET="your-unique-bucket-name"
export SERVICE_ACCOUNT="spark-runner@${PROJECT_ID}.iam.gserviceaccount.com"

export INPUT_URI="gs://${BUCKET}/transactions/"
export OUTPUT_URI="gs://${BUCKET}/analytics/hourly-transactions/"
export SCRIPT_URI="gs://${BUCKET}/jobs/hourly_transactions.py"

# 登录并选择项目后，启用所需 API。
gcloud config set project "${PROJECT_ID}"
gcloud services enable dataproc.googleapis.com storage.googleapis.com

# 上传作业脚本。
gcloud storage cp hourly_transactions.py "${SCRIPT_URI}"

# 提交托管 Spark 批任务，无需预先创建常驻集群。
gcloud dataproc batches submit pyspark "${SCRIPT_URI}" \
  --batch="hourly-transactions-$(date +%Y%m%d-%H%M%S)" \
  --region="${REGION}" \
  --service-account="${SERVICE_ACCOUNT}" \
  --deps-bucket="${BUCKET}" \
  -- \
  --input="${INPUT_URI}" \
  --output="${OUTPUT_URI}"

如果结果需要直接进入 BigQuery，可以在运行环境已提供 BigQuery Connector、服务账号也具有目标数据集写入权限的前提下，将输出部分改为：

TARGET_TABLE = "your-project.analytics.hourly_transactions"

(
    hourly.write
    .format("bigquery")
    .option("table", TARGET_TABLE)
    .mode("append")
    .save()
)

生产环境还应为输入模式增加显式校验，避免上游字段变化导致金额聚合错误；对重复执行的任务，则需要设计分区覆盖、幂等写入或基于批次标识的去重机制。

迁移时不要只统计“作业是否成功”

PayPal 报告的 25% 处理时间改善和 30% SLA 提升，是其特定工作负载与迁移方案的结果，不应直接当作其他团队的容量承诺。更稳妥的做法，是按作业族建立迁移基线：

迁移顺序也值得谨慎设计。可以先选择依赖较少、输出容易核对的批处理作业，建立提交、监控、权限和成本归集模板，再迁移关键 SLA 任务。直接把旧平台上的资源参数照搬到托管环境，往往会保留过度分区、海量小文件或不合理 Shuffle 等历史问题。

落地前的检查清单

• 按真实峰值数据量完成性能与并发测试，而不是只测试少量样本。
• 为 Cloud Storage、BigQuery 和 Spark 执行账号实施最小权限。
• 使用固定版本的运行时与依赖，避免升级造成不可预测的行为变化。
• 对高成本作业设置预算、标签、资源上限和异常告警。
• 同时验证结果正确性与执行速度，尤其是金融、风控和结算数据。
• 为区域故障、上游延迟、重试和重复写入设计恢复流程。
• 迁移后淘汰重复平台与旧链路，否则统一平台会再次变成新的技术孤岛。

PayPal 的案例说明，大规模分析现代化的价值不只是把集群启动时间从数月缩短到数分钟。真正的收益来自减少平台差异、压缩数据链路，并让工程师把时间从硬件供给和故障救火转向作业优化与业务实验。托管 Spark 提供了弹性执行基础，但是否能获得同样的组织收益，仍取决于数据治理、成本控制、可观测性和迁移纪律。