Spark Structured Streaming 如何读取不同凭证和 schema 的多个 S3 bucket？

2026-11-04技术老齐

结论

可以使用 Spark Structured Streaming。对于约 500 buckets × 3 entities 的场景，建议优先采用“外部发现新增文件 + 有界 work item + Spark 批量处理”的架构。这些 work item 可以交给 Structured Streaming 消费，也可以继续由现有 scheduler 驱动 micro-batch Spark 作业来处理。

也可以直接读取客户 bucket：为不同 bucket/prefix 创建文件流源，再按相同 entity 合并。这种方式仍需管理约 1,500 个文件源。即使合并成几个查询，凭证配置、文件发现和故障隔离仍要分别处理。

如果现有 progress tracker 可靠，延迟要求也允许按分钟处理，可以先保留当前方案。等到需要 Structured Streaming 的触发、checkpoint 或下游流式计算能力时，再引入它。

多个 bucket 能否组成一个逻辑 stream？

文件源和流查询是两个层次：

文件源负责发现某个路径下的新增文件。
流查询可以包含多个文件源，并合并它们的结果。
一个 Spark 应用、一个 SparkSession 可以运行多个流查询。

原生 PySpark DataStreamReader.parquet() 接收一个路径字符串，不能像批处理中的 spark.read.parquet(*paths) 那样传入多个 bucket，也不要用逗号拼接路径来表示 bucket 列表。DataStreamReader.parquet API

对于 schema 相同的 entity，可以分别建立流源，再合并：

from pyspark.sql import functions as F

# 仅为示例，生产环境应使用该 entity 的实际 schema。
entity_schema = "record_id STRING, value DOUBLE, date STRING"

def read_entity(customer_id, bucket):
    return (
        spark.readStream
        .schema(entity_schema)
        .option("maxFilesPerTrigger", 100)
        .parquet(f"s3a://{bucket}/entity_1/")
        .withColumn("customer_id", F.lit(customer_id))
    )

entity_stream = read_entity(
    "customer-a", "customer-a-bucket"
).unionByName(
    read_entity("customer-b", "customer-b-bucket")
)

这会得到一个包含两个文件源的 streaming DataFrame，调用 writeStream.start() 后才会启动查询。unionByName() 按列名合并，不会去重。unionByName API

这种方式适合客户较少、输入集合稳定的情况。从同一个 checkpoint 恢复查询时，不支持随意增减输入源。因此，如果把 500 个客户固定在一个 union 查询里，客户接入和退出就需要处理查询迁移。查询变更与恢复语义

不同客户的凭证如何配置？

使用 Hadoop S3A 时，应按 bucket 配置：

fs.s3a.bucket.<bucket-name>.<option>

例如，fs.s3a.access.key 对应的 bucket 级配置是：

fs.s3a.bucket.customer-a-bucket.access.key

在 Spark 配置中，还要加上 spark.hadoop. 前缀。S3A 会用 bucket 级配置覆盖全局配置，因此同一个应用可以访问使用不同凭证的 bucket。S3A 按 bucket 配置

如果客户提供长期 access key，可以将各客户的凭证保存到不同的 Hadoop Credential Provider 文件中，Spark 配置只引用文件位置：

spark.hadoop.fs.s3a.bucket.customer-a-bucket.aws.credentials.provider=org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider
spark.hadoop.fs.s3a.bucket.customer-a-bucket.security.credential.provider.path=jceks://file/etc/s3-secrets/customer-a.jceks

spark.hadoop.fs.s3a.bucket.customer-b-bucket.aws.credentials.provider=org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider
spark.hadoop.fs.s3a.bucket.customer-b-bucket.security.credential.provider.path=jceks://file/etc/s3-secrets/customer-b.jceks

示例假设每个 JCEKS 文件中已经保存了对应客户的 fs.s3a.access.key 和 fs.s3a.secret.key。配置路径本身不包含密钥。Hadoop Credential Providers

在 Kubernetes 上，还需要处理以下问题：

将所需凭证文件挂载到 driver 和 executor，确保两者都能访问。
在第一次访问 bucket 前完成配置。不要在客户循环中反复修改全局 fs.s3a.access.key，否则并发读取和已缓存的文件系统实例会让行为难以控制。
为临时凭证配置明确的刷新机制。仅将一组 STS 凭证放入配置，不会自动获得长期续期能力。
根据镜像实际包含的 Hadoop、hadoop-aws 和 AWS SDK 版本选择 provider，不能直接混用不同版本文档中的类名。S3A 认证文档

如果客户允许跨账号授权，也可以按 bucket 使用不同角色。采用这种方式前，需要分别确认工作负载身份、角色信任关系和 provider 的刷新支持。Kubernetes ServiceAccount 的身份本身并不保证它有权访问所有客户 bucket。S3A Assumed Roles

按 bucket 配置只解决凭证选择问题。如果客户之间要求严格的权限隔离，应拆成不同的 Spark 应用，或将客户划入受限分组，避免一个进程持有全部客户凭证。

应该按 entity 还是按客户拆分？

先确认同一个 entity 在不同客户中的 schema 是否一致。题目只说明三个 entity 的 schema 不同，没有保证同名 entity 在所有客户中都相同。

文件流源通常应显式指定 schema。对于现有的 entity_1/date=.../ 布局，可以读取 entity_1/ 根路径，让 Spark 发现 date=... 分区。需要使用的分区列也应包含在读取 schema 中。文件流的 schema 与分区发现

方案	适用条件	主要代价
每个客户 × entity 一个查询	需要独立进度和重试边界	约 1,500 个查询，运维负担大
每个 entity 一个 union 查询	同名 entity 的 schema 一致，客户集合稳定	每个查询仍有约 500 个文件源，故障影响范围较大
按客户分组，再按 entity 建查询	希望限制单个 driver 的负担	需要管理分组、资源和迁移
统一消费 work item，按 entity/schema 路由	客户多，接入变化频繁	文件发现、重试和幂等需要自行设计

不建议仅为减少 stream 数量，就把三个不同 entity 强行合并成大量可空列。即便输出 schema 能统一，各 entity 的输入 schema、处理规则和 schema 演进策略仍需分别定义。

查询隔离与应用隔离也有区别：两个查询放在同一个 driver 中，仍然共享 driver 生命周期和资源。

manifest/work-item 方案如何落地？

采用这个方案可以保留客户现有的 S3 布局。控制路径只保存待处理文件的元数据，Parquet 文件仍留在客户 bucket 中。

外部发现器通过轮询发现新增文件；如果允许配置通知，也可以使用对象事件。
按客户、entity 和 schema 版本生成大小受控的 work item。
将 work item 发布到可持久化、可重放的控制路径。
Spark 读取 work item，再通过普通 spark.read 读取指定的 Parquet 文件。
输出成功提交后，记录任务完成状态；失败任务可以重试。

一个 work item 可以包含：

{
  "work_id": "customer-a-entity-1-batch-000123",
  "customer_id": "customer-a",
  "entity": "entity_1",
  "schema_version": "v1",
  "base_path": "s3a://customer-a-bucket/entity_1/",
  "paths": [
    "s3a://customer-a-bucket/entity_1/date=2026-11-04/part-0001.parquet"
  ]
}

manifest 中应保存明确的文件清单。如果只保存一个会继续增长的目录，失败重试时读到的输入可能发生变化。

Structured Streaming 不会自动读取 manifest 中路径所指向的文件，这部分需要在 foreachBatch 中实现。以下是普通 PySpark-on-Kubernetes 应用的结构示例，其中 SCHEMAS、validate_work_item 和 write_idempotently 需要由业务代码提供：

from pyspark.sql import functions as F

work_schema = """
    work_id STRING,
    customer_id STRING,
    entity STRING,
    schema_version STRING,
    base_path STRING,
    paths ARRAY<STRING>
"""

work_stream = (
    spark.readStream
    .schema(work_schema)
    .option("maxFilesPerTrigger", 100)
    .json("s3a://ingestion-control/work-items/shard-00/")
)

def process_work_items(work_df, batch_id):
    for item in work_df.toLocalIterator():
        # 检查客户与 bucket/prefix 的对应关系、schema 版本和清单大小。
        validate_work_item(item)

        data = (
            spark.read
            .schema(SCHEMAS[(item.entity, item.schema_version)])
            .option("basePath", item.base_path)
            .parquet(*item.paths)
            .withColumn("customer_id", F.lit(item.customer_id))
            .withColumn("work_id", F.lit(item.work_id))
        )

        # 必须执行实际写入，并等待提交完成。
        # 重复调用同一个 work_id 不得产生重复结果。
        write_idempotently(data, work_id=item.work_id)

query = (
    work_stream.writeStream
    .foreachBatch(process_work_items)
    .option(
        "checkpointLocation",
        "s3a://ingestion-control/checkpoints/shard-00/"
    )
    .trigger(processingTime="30 seconds")
    .start()
)

query.awaitTermination()

读取明确的文件列表时，basePath 用于保留 date=... 分区列的解析上下文。控制 bucket 的凭证也需要单独配置。

100 个 manifest 文件和 30 seconds 都是示例值。这段代码逐个处理 work item，用于说明处理流程。生产环境可以将同一客户、entity 和 schema 的文件合并成有界批次，减少小任务。不要在 executor 的 map 或 foreachPartition 中再启动 Spark 读取作业。

manifest 方案将文件发现移到流查询之外，发现器仍需保证能可靠地发现新增对象。如果依靠轮询，仍会产生扫描成本。

checkpoint 不能替代哪些工作？

foreachBatch 默认只有至少一次写入语义。同一批次可能在故障后重新执行，因此业务写入必须支持幂等。foreachBatch 写入语义

需要处理输出已经成功、完成状态尚未写入时程序崩溃的情况。仅靠先写数据、再标记 done，无法避免重复。可以采用以下方式：

在同一事务中提交输出与处理记录。
以稳定的业务键执行幂等 upsert。
使用目标存储支持的提交协议，以 work_id 识别重复提交。

batch_id 可用于同一查询恢复时的批次去重。但外部重复发布的 work item 可能进入不同批次，因此仍应使用稳定的 work_id。foreachBatch API

还需要满足以下约束：

源文件发布完成后保持不可变，并保留到任务成功且必要的重试窗口结束。只记录对象版本，却按普通路径读取最新内容，无法保证重放一致。
每个查询使用独立、持久化的 checkpoint，不要保存在 Pod 临时磁盘中，也不要让两个活动查询共用一个目录。
一个 work item 失败会导致整个 micro-batch 失败。如果需要让其他客户继续推进，应对控制路径分片，或实现持久化的任务重试与隔离机制。
文件源 checkpoint 跟踪文件进度，不负责业务记录去重。相同数据出现在两个不同文件中时，仍可能被处理两次。

500 个 bucket 时如何选择？

仅凭 500 × 3 无法判断 Structured Streaming 能否承受。需要测量每轮文件发现耗时、历史对象数量、新增文件速率、文件大小、driver 负载，以及单个客户失败后的恢复时间。

直接维护大量文件流源时，maxFilesPerTrigger 限制的是各源每批选择的新文件数量。它不是整个应用的全局吞吐上限，也不能保证 S3 listing 成本受控。

对于这个场景，建议按以下顺序实施：

保留现有 progress tracker，将新增文件发现与 Spark 数据处理拆开。
引入有界 work item，明确输入清单、schema 版本和幂等提交规则。
按客户组和负载控制并发，避免一个 driver 管理全部客户。
如果外部 scheduler 已能满足延迟要求，继续使用批作业。
如果需要常驻消费、自动触发，或后续有跨批次聚合、watermark 等需求，再用 Structured Streaming 消费控制路径。

这样可以在保留客户 S3 布局的同时，逐步改造现有应用。是否引入 Structured Streaming，取决于调度和计算需求；文件发现与提交正确性仍需单独设计。