基于EMR Spark处理S3 10TB级日志数据集的性能优化问询

场景概述 #

每日需用EMR Spark读取S3中约10TB日志数据集，经分区列+非分区列过滤后写回S3，过滤后数据量小于1TB。源数据按day/hour/col1/col2分区，小时数据分布均匀，按天批量处理、小时拆分查询，示例代码如下：

for(date <- 1 to 30){
        var output_path = "s3a://dest-bucket/logs/day=%d/".format(date)
        spark.table("s3a://source-bucket/logs/day=%d/")
            .filter('col1.isin("A", "B", "C"))      // col1 is partitioned in source S3
            //.filter('col3.isin("X", "Y", "Z"))    --> Point 4
            .withColumn("hid", hash($"id") % 2000)  --> Point 5
            .select("col1","col2","col3","col4")
            // .repartition(col("hid"))             --> Point 5
            .repartition(col("hour"))               --> Point 1
            .write
            // .partitionBy("hid")
            .partitionBy("hour")                    --> Point 2
            .option("header","true").mode("overwrite").parquet(output_path)
}

当前任务每日耗时10-15小时，偶发Executor丢失失败；单节点AWS CLI全量复制耗时相近，期望通过大型集群提速。以下是具体优化疑问及解答：

优化疑问与解答 #

1. Spark(v2.4.8)是否需repartition(col("hour"))来识别源分区？对分区数据源重分区的利弊，以及针对hour列均匀分布的情况重分区是否有益？ #

• 不需要repartition(col("hour"))识别源分区：Spark读取分区表时会自动识别所有分区列（包括hour），无需手动重分区触发。
• 重分区的利：若原数据分区文件数量过少/过大，重分区能让任务并行度匹配集群资源，避免单Executor处理超大文件；按hour重分区后，写入阶段每个hour的数据已在对应Executor上，可减少局部shuffle。
• 重分区的弊：会触发额外shuffle操作，增加磁盘IO与网络开销——尤其源数据已按hour分区且分布均匀时，该shuffle完全冗余。
• 针对hour均匀分布的场景：不建议执行此重分区。源数据已按hour分片，Spark会自动并行读取，额外重分区只会增加无意义的shuffle成本。

2. 写入时用partitionBy("hour")可提升24倍速度，否则Spark会顺序写入，如何让Executor并行执行写入？ #

• 不用partitionBy导致顺序写入的核心原因是输出文件数量不足，或数据未合理拆分到多个Executor。解决方法：
  • 若无需按hour分区，通过repartition(N)指定足够多的分区数（N建议匹配集群总核数的2-3倍，比如等于spark.sql.shuffle.partitions值），让多个Executor同时写入不同文件。
  • 若过滤后数据量大幅减少，用coalesce(N)替代repartition（无shuffle开销），但需保证N足够支撑并行写入。
  • 调整spark.sql.files.maxPartitionBytes（默认128MB），拆分过大的源文件，提升读取阶段并行度，让更多Executor参与后续处理。

3. 写入时使用partitionBy，是否需要配合repartition？ #

• 分两种情况判断：
  • 源数据已按partitionBy指定列（如hour）分区：无需配合repartition。Spark读取时已按该列分片，写入时直接将分片数据对应到目标分区，无额外shuffle需求。
  • 源数据未按目标分区列分区，或目标分区列数据分布不均：可执行repartition(col("目标分区列"), N)，N为每个目标分区下的文件数，既保证每个分区的数据由多个Executor并行写入，又避免生成过多小文件。
• 你的场景中源数据已按hour分区，建议去掉repartition(col("hour"))，节省shuffle时间。

4. 若放弃非分区列col3的过滤条件，能否显著提升查询性能？ #

• 会有显著提升：
  • 原场景中filter('col3.isin("X", "Y", "Z"))属于列级别全表扫描，需读取每个分区内的所有文件、解析每行记录的col3值过滤，IO与计算成本极高。
  • 放弃该过滤后，Spark仅需根据col1的分区过滤（col1.isin("A", "B", "C")）直接读取对应分区数据，无需解析文件内容，读取速度会大幅提升——尤其源数据按col1分区时，可跳过大量无关分区。
• 若业务允许放弃该过滤，性能提升非常明显；若必须保留，建议将col3加入源数据分区键，或使用Parquet bloom filter等列式索引加速过滤。

5. 最终目标是基于用户ID哈希生成hid列并以此分区写入，但重shuffle导致集群失败，集群为200台m5.8xlarge节点（每台128GB内存），如何优化基于新非分区列的重shuffle？ #

• 优化方向如下：
  1. 调整shuffle配置：
     • 降低spark.sql.shuffle.partitions：当前设置为5995，200台m5.8xlarge（每台分配5核）总核数为1000，建议将shuffle partitions设为总核数的2-3倍（2000-3000），避免过多小分区导致资源碎片化。
     • 提高spark.reducer.maxBlocksInFlightPerAddress：当前为20，可适当提升至50，提升Reducer拉取数据效率，但需避免引发OOM。
     • 开启spark.sql.adaptive.enabled（Spark 2.4支持）：让Spark自动调整shuffle分区数，合并小分区，减少资源浪费。
  2. 优化内存与GC：
     • 当前Executor内存18GB，m5.8xlarge有128GB内存，可将spark.executor.memory提至60GB、spark.executor.memoryOverhead提至10GB，给shuffle操作更多内存空间，减少磁盘溢出。
     • 添加spark.executor.extraJavaOptions="-XX:+UseG1GC"，提升内存回收效率。
  3. 分步处理降低单次shuffle规模：
     • 先执行col1过滤减少数据量，再生成hid并shuffle——过滤后数据小于1TB，远小于源数据10TB，可大幅降低shuffle压力。
     • 按小时拆分任务，每个小时单独处理生成hid并写入，避免单次任务负载过高。
  4. 优化写入阶段：
     • 使用repartition(col("hid"), N)，N为每个hid分区下的文件数，平衡Executor写入压力。
     • 保持spark.hadoop.parquet.enable.summary-metadata=false（已开启），避免生成额外元数据文件，加快写入速度。
  5. 排查Executor丢失问题：
     • 当前spark.executor.heartbeatInterval=3600设置过大，易导致Executor因心跳超时被判定为丢失，建议改回默认10s或调整为60s，同时确保集群网络稳定。

当前Spark配置优化建议 #

针对200台m5.8xlarge集群规模，可调整配置如下：

spark-shell  \
--master yarn  \
--conf "spark.executor.instances=1200"  # 每台机器跑6个Executor（5核*6=30核，留2核给系统），200*6=1200
--conf "spark.default.parallelism=3600"  # 总核数1200*5=6000，设置为总核数的0.6倍
--conf "spark.sql.shuffle.partitions=3600"  # 匹配default parallelism
--conf "spark.driver.cores=5" \
--conf "spark.driver.maxResultSize=20g" \
--conf "spark.driver.memory=18g" \
--conf "spark.driver.memoryOverhead=3g" \
--conf "spark.executor.cores=5" \
--conf "spark.executor.memory=60g" \
--conf "spark.executor.memoryOverhead=10g" \
--conf "spark.executor.heartbeatInterval=60" \
--conf "spark.hadoop.orc.overwrite.output.file=true" \
--conf "spark.hadoop.parquet.enable.summary-metadata=false" \
--conf "spark.sql.execution.arrow.pyspark.enabled=true" \
--conf "spark.sql.execution.arrow.pyspark.fallback.enabled=true" \
--conf "spark.sql.parquet.mergeSchema=false" \
--conf "spark.sql.parquet.int96RebaseModeInRead=CORRECTED" \
--conf "spark.sql.debug.maxToStringFields=100" \
--conf "spark.hadoop.fs.s3.connection.maximum=1000" \
--conf "spark.hadoop.fs.s3.connection.timeout=300000" \
--conf "spark.hadoop.fs.s3.threads.core=250" \
--conf "spark.hadoop.fs.s3a.connection.maximum=1000" \
--conf "spark.hadoop.fs.s3a.connection.timeout=300000" \
--conf "spark.hadoop.fs.s3a.threads.core=250"\
--conf "spark.reducer.maxBlocksInFlightPerAddress=50" \
--conf "spark.sql.adaptive.enabled=true" \
--conf "spark.sql.adaptive.shuffle.targetPostShuffleInputSize=64m"

内容的提问来源于stack exchange，提问作者Hossein Mousavi