Сравните две PCollections для удаления

Question

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

Не могли бы вы предложить лучший способ сделать этот сценарий с использованием конвейера потока данных (предпочтительно Java).

Я определил одну вещь: используя функцию upsert в CloudSQL, мы могли бы выполнять вставку / обновление записей с помощью jdbc.JdbcIO. Но я не знаю, как определить коллекцию для удаления.

Answer 1

Вы можете прочитать старую и новую таблицы и выполнить Join , за которым следует DoFn, который сравнивает два и только выводит измененные элементы, которые затем могут быть записаны где угодно.