Databricks 到 Cosmos 之间的 UPSERT /INSERT/UPDATE

Question

目前我们使用 Azure Databricks 作为转换层，转换后的数据通过连接器加载到 Cosmos DB。

场景：

我们有 2 个文件作为源文件。

第一个文件包含名称、年龄

第二个文件包含名称、州、国家

在 Cosmos 中，我使用 id、分区键创建了集合

在 databricks 中，我将这 2 个文件作为 Dataframe 加载并创建一个临时表来查询内容。

我正在从第一个文件中查询内容 [从文件中选择名称作为 id、名称、年龄]并将其加载到 Cosmos 集合中。

来自第二个文件。我正在使用 [select name as id, state, country] 并加载到同一个集合，期望第二个文件中的内容根据 id 字段插入到同一文档的同一集合中。

这里的问题是，当我从第二个文件加载内容时，第一个文件中的“年龄”属性被删除，在 cosmos 文档中只看到 id、name、state、country。发生这种情况是因为我在 databricks 中使用 UPSERT 加载到 Cosmos。

当我将 UPSERT 更改为 INSERT 或 UPDATE 时，它会抛出错误，提示“具有 id 的资源已存在”

Databricks 与 Cosmos 的连接：

val configMap = Map(
  "Endpoint" -> {"https://"},
  "Masterkey" -> {""},
  "Database" -> {"ods"},
  "Collection" -> {"tval"},
  "preferredRegions" -> {"West US"},
  "upsert" -> {"true"}) 
  val config = com.microsoft.azure.cosmosdb.spark.config.Config(configMap)

有没有办法从第二个文件中插入属性而不删除已经存在的属性。我没有使用 JOIN 操作，因为用例不适合使用。

Answer 1

从模糊的记忆中，您需要在数据框中设置 id 属性以匹配两个数据集。 如果您省略此字段，Cosmos 会生成一条新记录 - 这就是您正在发生的事情。

因此，如果 df1 & df2 在第一条记录上的 id=1，那么第一个将插入它，第二个将更新它。

但如果它们是相同的记录，那么加入 Spark 会更有效率。