Spark Structured Streaming中S3文件重复扫描的原因与优化咨询

问题原因分析 #

• 两次CSVScan是因为你基于同一数据源创建了两个独立的过滤分支，Spark默认会为每个分支发起单独的文件扫描任务，不会自动复用已读取的文件内容。
• InMemoryFileIndex只是把文件的路径、元数据等索引信息存在内存，不是文件的实际内容，所以每个分支都会重新从S3拉取完整文件，导致读取量翻倍。
• 内存复用不会自动发生：如果没有显式缓存，Spark不会将读入的文件数据保留在内存中供后续分支复用，每次扫描都是独立的IO操作。

优化方案 #

1. 显式缓存原始DataFrame

在读取CSV文件后，先调用cache()（或persist()指定存储级别）缓存原始数据，之后所有分支基于缓存后的DataFrame做过滤，这样只需要读取一次S3文件：

// 读取S3文件后立即缓存
val rawDF = spark.read.csv(s3FilePath).cache()

// 基于缓存后的DataFrame创建两个分支
val branch1DF = rawDF.filter("你的自定义SQL过滤表达式1")
val branch2DF = rawDF.filter("你的自定义SQL过滤表达式2")

注意：结构化流场景下，要结合作业的检查点配置，确保缓存的内存占用在可控范围内，避免内存泄漏。可以根据数据量选择合适的存储级别，比如MEMORY_AND_DISK防止内存不足时数据丢失。

2. 先打标签再拆分分支（业务适配前提下）

如果业务逻辑允许，可以先给每行数据标记所属分支，再拆分，这样全程只需要一次文件扫描：

import org.apache.spark.sql.functions.{when, expr}

val taggedDF = rawDF.withColumn(
  "branch_tag",
  when(expr("表达式1"), "branch1")
    .when(expr("表达式2"), "branch2")
)

val branch1DF = taggedDF.filter("branch_tag = 'branch1'")
val branch2DF = taggedDF.filter("branch_tag = 'branch2'")

3. 开启S3文件本地缓存

启用Spark的S3本地缓存功能，让同一Executor上的任务重复读取同一文件时，直接用本地缓存的副本，避免重复从S3下载：

// 在SparkSession初始化时配置
val spark = SparkSession.builder()
  .config("spark.sql.s3.cache.enabled", "true")
  .config("spark.sql.s3.cache.dir", "/local/cache/path") // 可选，指定本地缓存目录
  .getOrCreate()

如果用的是自定义S3-SQS连接器，也可以检查连接器是否有类似的本地缓存配置项，进一步优化IO。

4. 调整过滤表达式以适配优化

尽量用Spark内置函数组合过滤逻辑，避免过于复杂的自定义表达式，让Spark优化器能更好地处理过滤逻辑，即使无法下推到S3 Select，也能在数据加载后尽早过滤，减少后续处理的数据量。

内容的提问来源于stack exchange，提问作者Umang Desai