我有一個如下的資料框：
val data = Seq(("James", "Sales", 34, "Developer"), ("Michael", "Sales", 56, "Architect"), ("Robert", "Sales", 30, "Manager"), ("Maria", "Finance", 24, "Consultant"))
val df1 = data.toDF("name","dept","id", "role")
df1.printSchema()
root
|-- name: string (nullable = true)
|-- dept: string (nullable = true)
|-- id: integer (nullable = true)
|-- role: string (nullable = true)
我有一個具有相同列和確切架構的配置單元表：
val df2 = spark.sql("select * from db.table")
從傳入的資料框中，df1我得到了兩條新記錄和兩條更新記錄。
val df2 = spark.sql("select * from db.table where name in ('James', 'Michael')")
df2.show()
------- ------- --- ----------
| name| dept| id| role|
------- ------- --- ----------
| James| Sales| 34| Associate|
|Michael| Sales| 56| Junior|
------- ------- --- ----------
這里使用的鍵是：dept&id
在我之前的一個專案中，我們曾經將傳入的資料幀與我們的 Hive 表的磁區連接到我們的臨時表中，并簡單地運行exchange partition以將現有的 hive 磁區與包含合并資料的臨時表交換。
我們正在使用 Spark 的 Databricks 發行版。我們的 hive 表建立在 Databricks delta 湖上，有數百萬行。有沒有其他方法可以將傳入的資料幀 df1 與配置單元表合并？如果是這樣，我怎樣才能在不影響性能的情況下實作它。
uj5u.com熱心網友回復：
正如 Tim 所提到的，如果您的目標表已經在 Delta 上，那么您只需要使用MERGE INTOSQL 命令或相應的 Scala API（請參閱Delta Merge 上的檔案）。你需要這樣的東西：
import io.delta.tables._
import org.apache.spark.sql.functions._
DeltaTable.forName(spark, "db.table")
.as("target")
.merge(
df1.as("updates"),
"target.dept = updates.dept and target.id = updates.id")
.whenMatched
.updateAll()
.whenNotMatched
.insertAll()
.execute()
不匹配的資料將按原樣插入，匹配的資料將放入包含原始記錄的重寫檔案中。通常重寫是主要的性能損失，您可能需要減小檔案大小以重寫更少的資料（請參閱檔案） - 在較新的版本中，可以這樣配置表，以便 Databricks Spark 引擎將自動找到最佳檔案大小以減少在不影響讀取模式的情況下重寫時間（參見檔案）
轉載請註明出處，本文鏈接：https://www.uj5u.com/qianduan/333882.html