Блог

Я выполняю операцию слияния в дельта-таблице databricks, как показано ниже —

 spark.sql(""" MERGE INTO <delta table name> deltatbl USING <temp view> source
           ON   deltatbl.col1   =    source.col1
           AND  deltatbl.col2   =    source.col2
           WHEN NOT MATCHED THEN INSERT
           (col1,col2) VALUES(source.Col1,source.Col2)  """)
 

Приведенный выше запрос вставляет дубликаты записей, несмотря на совпадение уникальных ключей. Как я могу добиться вывода, в который вставляются только не совпадающие записи. все столбцы являются частью ключа.

Из того, что я вижу, одна из проблем может заключаться в том, что то, что вы вставляете, содержит дубликаты. ЕСЛИ ОНИ НЕ СОВПАДАЮТ, ВСТАВКА не выполняет дедупликацию записей перед вставкой.

Чтобы не загружать дубликаты в таблицу, вы должны выполнить дедупликацию перед запуском слияния.

Вы можете либо сделать это с помощью API python:

 spark.table('{your_view_name_here}').dropDuplicates()
 

Или через SQL API:

 SELECT DISTINCT(*)
FROM {your_view_name_here}
 

без какой-либо дополнительной информации, которая является лучшим предположением, которое я могу предоставить.

Если вы хотите обновить существующие записи:

 MERGE INTO events
USING updates
ON events.eventId = updates.eventId
WHEN MATCHED THEN
  UPDATE SET events.data = updates.data
WHEN NOT MATCHED
  THEN INSERT (date, eventId, data) VALUES (date, eventId, data)
 

Если вы хотите вставить только для несуществующих записей : Обновите с тем же значением

 MERGE INTO events
USING updates
ON events.eventId = updates.eventId
WHEN MATCHED THEN
UPDATE SET events.data = events.data
WHEN NOT MATCHED
THEN INSERT (date, eventId, data) VALUES (date, eventId, data)
 

Ваше дело ,

 MERGE INTO <delta table name> deltatbl USING <temp view> source
       ON   deltatbl.col1   =    source.col1
       AND  deltatbl.col2   =    source.col2
       WHEN MATCHED THEN
         UPDATE SET deltatbl.data = deltatbl.data
       WHEN NOT MATCHED THEN INSERT
       (col1,col2) VALUES(source.Col1,source.Col2)