使用 R 的 Databricks Delta Table Merge 语句答案

【问题标题】：Databricks Delta Table Merge statement using R使用 R 的 Databricks Delta Table Merge 语句
【发布时间】：2022-01-02 00:41:37
【问题描述】：

我最近开始研究 Databricks，我一直在尝试找到一种方法来在 Delta 表上执行合并语句，尽管使用的是 R api（最好是 sparklyr）。最终目的是以某种方式施加“重复”约束，如here 所述。上述文档描述了 Python 工作流程：

deltaTable.alias("logs").merge(
    newDedupedLogs.alias("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId") \
  .whenNotMatchedInsertAll() \
  .execute()

但是，我想知道是否有一种直接的方法可以通过 R 实现这一目标。由于我是新用户（如上所述），因此我们将非常感谢任何关于此事的帮助/想法。提前致谢！

【问题讨论】：

标签： r duplicates databricks sparklyr delta-lake

【解决方案1】：

提供此答案，因为您评论说没有 R Delta Lake API 支持。现在有一个为 Delta Lake 提供 R API 的新 R 包：dlt。语法与 Delta Lake 的 Python API 非常相似。

在你的例子中：

# Install and laod the `dlt` package
remotes::install_gitlab("zero323/dlt")
library(dlt)
...

# Use the Delta Lake R API from the dlt package
deltaTable <- dlt_for_path("<path to table>")

deltaTable %>%
  dlt_alias("logs") %>%
  dlt_merge(alias(newDedupedLogs, "newDedupedLogs"), expr("newDedupedLogs.uniqueId = logs.uniqueId")) %>%
  dlt_when_not_matched_insert_all() %>%
  dlt_execute()

【讨论】：

【解决方案2】：

Delta Lake 没有官方的 R API，但您可以使用 SQL 的 MERGE INTO 命令（我经常在 Scala/Python 中这样做，只是因为它更易于阅读，至少对我而言）。您需要使用要放入目标表的数据注册一个临时视图，然后运行sql，如下所示（为了便于阅读，字符串被拆分）：

library(SparkR)
updates_df <- ...get updates...
createOrReplaceTempView(updates, "updates")
result <- sql(
  "MERGE INTO <your_table> AS target USING updates 
   ON target.id = updates.id WHEN NOT MATCHED THEN INSERT *")

【讨论】：

很遗憾，没有官方支持 R Delta Lake API 支持。无论如何，非常感谢您提供的解决方案和反馈；任何情况下我都可以避免使用 TempView？我想不是，对吗？
理论上可以写入磁盘并使用，但会比temp view差
顺便说一句，看起来 R 的一些包装器已经发布，虽然不是官方的