如何从databricks中的增量文件中删除数据？

Question

我想从 databricks 中的增量文件中删除数据。 我正在使用这些命令
例如：

PR=spark.read.format('delta').options(header=True).load('/mnt/landing/Base_Tables/EventHistory/')
PR.write.format("delta").mode('overwrite').saveAsTable('PR')
spark.sql('delete from PR where PR_Number=4600')

这是从表中删除数据，但不是从实际的增量文件中删除数据。我想删除文件中的数据而不使用合并操作，因为连接条件不匹配。谁能帮我解决这个问题。

谢谢

Answer 1

请记住：Delta 的 DELETE 不支持子查询。

问题链接：https://github.com/delta-io/delta/issues/730

从文档本身来看，替代方案如下

例如：

DELETE FROM tdelta.productreferencedby_delta 
WHERE  id IN (SELECT KEY 
              FROM   delta.productreferencedby_delta_dup_keys) 
       AND srcloaddate <= '2020-04-15'

如果是DELTA，可以写成如下

MERGE INTO delta.productreferencedby_delta AS d 
using (SELECT KEY FROM   tdatamodel_delta.productreferencedby_delta_dup_keys) AS k 
ON d.id = k.KEY 
  AND d.srcloaddate <= '2020-04-15' 
WHEN MATCHED THEN DELETE 

Answer 2

在 python 中使用 Spark SQL 函数将是：

dt_path = "/mnt/landing/Base_Tables/EventHistory/"
my_dt = DeltaTable.forPath(spark, dt_path)
seq_keys = ["4600"] // You could add here as many as you want
my_dt.delete(col("PR_Number").isin(seq_keys))

在 scala 中：

val dt_path = "/mnt/landing/Base_Tables/EventHistory/"
val my_dt : DeltaTable = DeltaTable.forPath(spark, dt_path)
val seq_keys = Seq("4600") // You could add here as many as you want
my_dt.delete(col("PR_Number").isin(seq_keys:_*))

Answer 3

您可以从 Delta 表中删除与谓词匹配的数据 https://docs.delta.io/latest/delta-update.html#delete-from-a-table

Answer 4

效果就像

delete from delta.`/mnt/landing/Base_Tables/EventHistory/` where PR_Number=4600

Answer 5

如果您还没有弄清楚，您需要对表运行

VACCUM

操作才能删除物理数据文件。只需执行

DELETE

操作，然后执行

VACCUM

操作就足以满足您的用例。