Spark Partitionby 未按预期扩展

Question

输入：

输入数据集包含存储为 parquet 的多个文件中的 1000 万个事务。包括所有文件在内的整个数据集的大小在 6 到 8GB 之间。

问题说明：

根据客户 ID 对交易进行分区，这将为每个客户 ID 创建一个文件夹，每个文件夹包含该特定客户完成的所有交易。

HDFS 对根目录中可以创建的子目录数量有 640 万个硬性限制，因此使用客户 ID 的最后两位数字（范围从 00,01,02...到 99）来创建顶部级别目录和每个顶级目录将包含所有以该特定两位数字结尾的客户 ID。

示例输出目录结构：

00/cust_id=100900/part1.csv
00/cust_id=100800/part33.csv

01/cust_id=100801/part1.csv
03/cust_id=100803/part1.csv

代码：

// Reading input file and storing in cache
val parquetReader = sparksession.read
  .parquet("/inputs")
  .persist(StorageLevel.MEMORY_ONLY) //No spill will occur has enough memory

// Logic to partition
var customerIdEndingPattern = 0
while (cardAccountEndingPattern < 100) {
  var idEndPattern = customerIdEndingPattern + ""
  if (customerIdEndingPattern < 10) {
    idEndPattern = "0" + customerIdEndingPattern
  }

  parquetReader
    .filter(col("customer_id").endsWith(idEndPattern))
    .repartition(945, col("customer_id"))
    .write
    .partitionBy("customer_id")
    .option("header", "true")
    .mode("append")
    .csv("/" + idEndPattern)
  customerIdEndingPattern = customerIdEndingPattern + 1
}

Spark 配置： Amazon EMR 5.29.0（Spark 2.4.4 和 Hadoop 2.8.5）

1 个主节点和 10 个从节点，每个节点都有 96 个 vCore 和 768GB RAM（Amazon AWS R5.24xlarge 实例）。硬盘是 EBS，在 30 分钟内达到 3000 IOPS。

            'spark.hadoop.dfs.replication': '3',
            'spark.driver.cores':'5',
            'spark.driver.memory':'32g',
            'spark.executor.instances': '189',
            'spark.executor.memory': '32g',
            'spark.executor.cores': '5',
            'spark.executor.memoryOverhead':'8192',
            'spark.driver.memoryOverhead':'8192',
            'spark.default.parallelism':'945',
            'spark.sql.shuffle.partitions' :'945',
            'spark.serializer':'org.apache.spark.serializer.KryoSerializer',
            'spark.dynamicAllocation.enabled': 'false',
            'spark.memory.fraction':'0.8',
            'spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version':'2',
            'spark.memory.storageFraction':'0.2',
            'spark.task.maxFailures': '6',
            'spark.driver.extraJavaOptions': '-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -XX:ConcGCThreads=12 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError="kill -9 %p"
            'spark.executor.extraJavaOptions': '-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -XX:ConcGCThreads=12 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError="kill -9 %p"

缩放问题：

1. 实验了从 10 个到 40 个从属设备（相应地调整 spark 配置），但结果仍然相同，该作业需要 2 多个小时才能完成（如第一张图片所示，每个作业需要一分钟以上while 循环运行 99 次）。此外，来自远程执行程序的读取几乎不存在（这很好）大多数是本地进程。

2. 分区似乎工作正常（参考第二张图片）每个实例有 5 个 RDD 块，并且始终运行 5 个任务（每个实例有 5 个内核，每个从节点有 19 个实例）。 GC 也进行了优化。

3. 在 while 循环中写入的每个 partitionby 任务需要一分钟或更长时间才能完成。

指标：

我们总共有 99 个工作的几个工作的样本持续时间

分区似乎没问题

1 个作业的总结，基本上是一个分区执行


完整作业完成后的几个实例的摘要，因此 RDD 块为零，第一行是驱动程序。

所以问题是如何进一步优化它，为什么它没有扩大规模？有没有更好的方法来解决它？我已经达到最高性能了吗？假设我可以在硬件方面获得更多资源，有什么我可以做得更好的吗？欢迎提出任何建议。

Answer 1

触摸每条记录 100 次是非常低效的，即使数据可以缓存在内存中并且不会被下游驱逐。更不用说单独坚持是昂贵的

您可以添加一个虚拟列

import org.apache.spark.sql.functions.substring

val df = sparksession.read
  .parquet("/inputs")
  .withColumn("partition_id", substring($"customer_id", -2, 2))

稍后用于分区

df
  .write
  .partitionBy("partition_id", "customer_id")
  .option("header", "true")
  .mode("append")
  .csv("/")

到avoid to many small files你可以先用更长的后缀重新分区

val nParts: Int = ???
val suffixLength: Int = ???  // >= suffix length used for write partitions

df
  .repartitionByRange(
    nParts,
    substring($"customer_id", -suffixLength, suffixLength)
  .write
  .partitionBy("partition_id", "customer_id")
  .option("header", "true")
  .mode("append")
  .csv("/")

此类更改将允许您一次处理所有数据，而无需任何显式缓存。