我正在嘗試根據 databricks 檔案將包含增量資料的資料框合并到我的基表中。
base_delta.alias('base') \
.merge(source=kafka_df.alias('inc'),
condition='base.key1=ic.key1 and base.key2=inc.key2') \
.whenMatchedUpdateAll() \
.whenNotMatchedInsertAll() \
.execute()
上述操作運行良好，但正如預期的那樣需要很多時間，因為正在掃描大量不需要的磁區。我在這里遇到了一個 databricks 檔案，一個帶有指定磁區的合并查詢。
該鏈接中的代碼：
spark.sql(s"""
|MERGE INTO $targetTableName
|USING $updatesTableName
|ON $targetTableName.par IN (1,0) AND $targetTableName.id = $updatesTableName.id
|WHEN MATCHED THEN
| UPDATE SET $targetTableName.ts = $updatesTableName.ts
|WHEN NOT MATCHED THEN
| INSERT (id, par, ts) VALUES ($updatesTableName.id, $updatesTableName.par, $updatesTableName.ts)
""".stripMargin)
磁區在IN條件中指定為1,2,3...但是在我的情況下，表首先按COUNTRY值磁區USA, UK, NL, FR, IND，然后每個國家/地區都有磁區YYYY-MM例如：2020-01, 2020-02, 2020-03
如果我有上面提到的嵌套結構，如何指定磁區值？非常感謝任何幫助。
uj5u.com熱心網友回復：
是的，您可以這樣做并且非常推薦，因為 Delta Lake 需要掃描與ON條件匹配的所有資料。如果您使用的是 Python API，您只需要使用正確的 SQL 運算式 as condition，并且您可以對磁區列進行限制，在您的情況下是這樣的（date是更新日期的列）：
base.country = 'country1' and base.date = inc.date and
base.key1=inc.key1 and base.key2=inc.key2
如果您有多個國家/地區，則可以使用，但在更新資料框內使用并匹配IN ('country1', 'country2')會更容易countrybase.country = inc.country
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/429922.html
上一篇：更新資料塊上的jar作業