Как найти разницу двух данных

Question

Я работаю над кодом Spark для модульного тестирования, который должен быть в состоянии генерировать разницу между двумя кадрами данных (сырое ведро и курированное ведро).Оба кадра данных (сегменты) одинаковы, и мы хотим выполнить этот код, чтобы зафиксировать возможные изменения после того, как мы скопируем файлы из raw в curated.,Мне известно, что я могу использовать функцию except следующим образом:

val difference =CuratedDataFrame.union(RawDataFrame).except(CuratedDataFrame.intersect(RawDataFrame))

+-----------+-------+-------------+---------+---------------+
|record     |pid    |feetype      |freq     |default        |
+-----------+-------+-------------+---------+---------------+
|          1|     45|          FAC|        Y|              T| 
|          1|     45|          FAC|        Y|            TTY| 
|          1|     47|          FAC|        R|              M| 
|          1|     99|          FAC|        R|              M|
+-----------+-------+-------------+---------+---------------+

Функция исключения возвращает всю строку, но мой желаемый результат выглядит следующим образом:

+-----------+-------+-------------+---------+---------------+
|record     |pid    |feetype      |freq     |default        |
+-----------+-------+-------------+---------+---------------+
|       null|[47,99]|         null|     null|         null  |
|       null|   null|         null|     null|       [T, TTY]|
+----------+-----------+-------+-------------+---------+-----

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

Для этого я использую следующий подход:

 val mapDiffs=(name: String) => when($"l.$name" === $"r.$name", null )
      .otherwise(array($"l.$name", $"r.$name")).as(name)
    val result = difference.as("l")
        .join(RawDataFrame.as("r"), $"l.primaryKey" === $"r.primaryKey","inner")
        .select($"l.primaryKey" :: cols.map(mapDiffs): _*)

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

+-----------+-------+-------------+---------+---------------+----------+
|record     |pid    |feetype      |freq     |default        |primaryKey|
+-----------+-------+-------------+---------+---------------+----------+
|          1|     40|          FAC|        A|              N|    FAC40A|
|          1|     45|          FAC|        Y|              T|    FAC45Y|
|          1|     47|          FAC|        R|              M|    FAC47R|
+-----------+-------+-------------+---------+---------------+----------+

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