Databricks读取Azure Blob全量Parquet过慢的优化策略咨询

一、存储层优化：重构第二组数据的存储格式 #

• 转用Delta Lake格式
  将第二组Parquet文件转换成Delta Lake表，利用它的分区索引、数据跳过能力，同时Spark读取Delta表时会自动按分区生成并行Task，解决单作业无并行度的问题。可以直接把原有月/日/小时的文件夹结构映射为Delta表的分区字段，一次性初始化后后续增量同步即可：

  // 初始化Delta表（仅需执行一次）
  spark.read.parquet("wasbs://container@storageacc.blob.core.windows.net/second-group-root")
    .write
    .partitionBy("month", "day", "hour")
    .format("delta")
    .save("wasbs://container@storageacc.blob.core.windows.net/second-group-delta")
  

  后续读取全量时直接读Delta表，并行度会显著提升。

• 升级到ADLS Gen2
  如果当前用的是标准Azure Blob存储，升级到ADLS Gen2。它支持HDFS风格的目录优化，Spark遍历大量子文件夹时的文件列表扫描速度会快很多，减少读取前的准备时间。

二、Spark读取阶段优化 #

• 显式声明分区字段
  不要依赖Spark自动推断分区，手动从文件路径解析出月/日/小时分区字段，让Spark按每个分区文件夹生成独立Task，强制并行读取：

  import org.apache.spark.sql.types.IntegerType
  import org.apache.spark.sql.functions.split, org.apache.spark.sql.functions.input_file_name
  
  val secondGroupDF = spark.read
    .option("basePath", "wasbs://container@storageacc.blob.core.windows.net/second-group-root")
    .parquet("wasbs://container@storageacc.blob.core.windows.net/second-group-root/*/*/*")
    .withColumn("month", split(input_file_name(), "/").getItem(6).cast(IntegerType)) // 索引值根据实际路径调整
    .withColumn("day", split(input_file_name(), "/").getItem(7).cast(IntegerType))
    .withColumn("hour", split(input_file_name(), "/").getItem(8).cast(IntegerType))
  

  这样每个小时的文件夹会对应一个Task，直接解决单作业无并行的问题。

• 调整并行度参数
  根据集群资源和数据量，手动设置Spark的并行度参数，确保有足够多的Task同时处理数据：

  // 调整shuffle分区数（默认200，可根据数据量增减）
  spark.conf.set("spark.sql.shuffle.partitions", "200")
  // 设置默认并行度
  spark.conf.set("spark.default.parallelism", "200")
  // 调整每个Task处理的数据量（建议128MB-256MB）
  spark.conf.set("spark.sql.files.maxPartitionBytes", "268435456") // 256MB
  

三、预处理与缓存优化 #

• 构建物化视图
  如果关联只用到第二组数据的部分字段或特定维度，提前构建物化视图，只保留关联需要的数据，减少读取和关联的数据量。比如每天同步增量数据到视图，全量读取时直接用视图，避免读取原始全量文件。

• 增量缓存全量数据
  利用Spark的缓存机制，每次调度前检查第二组是否有新数据，有则重新加载并缓存，否则直接复用缓存的数据：

  // 自定义函数：检查第二组是否有新数据（对比上次同步时间与最新文件夹修改时间）
  def hasNewData(): Boolean = {
    // 实现逻辑：调用Azure Blob API获取最新文件夹的修改时间，和上次同步时间对比
    true // 示例返回值
  }
  
  if (hasNewData()) {
    val secondGroupDF = spark.read.parquet("wasbs://container@storageacc.blob.core.windows.net/second-group-root").cache()
    secondGroupDF.createOrReplaceTempView("second_group_full")
  }
  
  // 关联时直接使用缓存的临时视图
  val resultDF = spark.sql("SELECT * FROM latest_first_group JOIN second_group_full ON your_join_condition")
  

四、关联阶段优化 #

• 广播小表（如果适用）
  如果第二组全量数据压缩后能放进Driver内存（比如小于10GB），用广播join避免shuffle，大幅提升关联速度：

  import org.apache.spark.sql.functions.broadcast
  val resultDF = latestFirstGroupDF.join(broadcast(secondGroupFullDF), "join_key")
  

  注意：数据量过大时不要用，会导致OOM。

• 指定合适的join策略
  根据两组数据的大小调整join策略：

  • 第二组数据远大于第一组：用shuffle hash join，禁用默认的sort merge join
  • 两组数据量级相当：保留sort merge join
    设置方式：

  // 禁用sort merge join，启用shuffle hash join
  spark.conf.set("spark.sql.join.preferSortMergeJoin", "false")
  

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