Spark delta Lake 合并上的分区修剪答案

【问题标题】：Partition pruning on Spark delta lake mergeSpark delta Lake 合并上的分区修剪
【发布时间】：2020-03-09 19:06:28
【问题描述】：

我正在使用 delta Lake ("io.delta" %% "delta-core" % "0.4.0") 并合并到 foreachBatch 中，例如：

foreachBatch { (s, batchid) =>
        deltaTable.alias("t")
          .merge(
            s.as("s"),
            "s.eventid = t.eventid and t.categories in ('a1', 'a2')")
          .whenMatched("s.eventtime < t.eventtime").updateAll()
          .whenNotMatched().insertAll()
          .execute()
      }

增量表按类别分区。如果我在（'a1'，'a2'）'中添加像'and t.categories'这样的分区过滤器，从火花图中我可以看到输入不是整个表。我认为它做了分区修剪。但是，如果我这样做： “s.eventid = t.eventid and t.categories=s.categories”，它仍然从增量表中加载所有数据。我希望它可以自动感知应该去哪些分区进行连接，有点下推。是否可以在不指定特定分区值的情况下进行分区修剪？我也尝试添加 ("spark.databricks.optimizer.dynamicPartitionPruning","true") 但不起作用。

谢谢

【问题讨论】：

我在 0.5 版中也面临同样的问题。
动态分区修剪仅适用于 databricks 运行时 5.5 lts

标签： apache-spark delta-lake

【解决方案1】：

你可以通过两种方式传递它。一种是传递值的静态方式，另一种是在合并语句中动态设置分区。

传递分区值的静态方式。

val categoriesList = List("a1", "a2")  
val catergoryPartitionList  = categoriesList.mkString("','")

foreachBatch { (s, batchid) =>
    deltaTable.alias("t")
      .merge(
        s.as("s"),
        "s.eventid = t.eventid and t.categories in ('$catergoryPartitionList')")
      .whenMatched("s.eventtime < t.eventtime").updateAll()
      .whenNotMatched().insertAll()
      .execute()
  }

将类别传递给 Merge 语句的动态方式如下：

val selectedCategories = deltaTable.select("categories").dropDuplicates()
  
val categoriesList = selectedCategories.map(_.getString(0)).collect()

val catergoryPartitionList  = categoriesList.mkString("','")

foreachBatch { (s, batchid) =>
    deltaTable.alias("t")
      .merge(
        s.as("s"),
        "s.eventid = t.eventid and t.categories in ('$catergoryPartitionList')")
      .whenMatched("s.eventtime < t.eventtime").updateAll()
      .whenNotMatched().insertAll()
      .execute()
  }

【讨论】：