Проблема с дублирующимися записями в базе данных во время процесса ETL - PullRequest
0 голосов
/ 10 мая 2018

Моя текущая архитектура ETL выглядит следующим образом:

s3 --> staging table --> python dataframe --> destination table
  1. Записи из s3 загружаются в промежуточную таблицу
  2. Сценарий Python подключен к промежуточной таблице
  3. Скрипт Python запускается каждый час для выполнения сложных преобразований
  4. Результирующий кадр данных из python загружается в таблицу назначения

Однако у меня возникла проблема с дублирующимися записями втаблица назначения:

| Time | New records (S3) | Redshift staging table (postgre) | Python DataFrame | Redshift Destination Table (postgre) | Duplicate records |
|------|------------------|----------------------------------|------------------|--------------------------------------|-------------------|
| 9am  | 3 records        | 3 records                        | 3 records        | 3 records                            | 0 (3-3)           |
| 10am | 2 records        | 5 (3+2) records                  | 5 records        | 8 (3+5) records                      | 3 (8-5)           |
| 11am | 4 records        | 9 (5+4) records                  | 9 records        | 17 (9+8) records                     | 8 (17-9)          |

Таким образом, в 11:00 в промежуточной таблице было 9 записей, но таблица назначения на 17 записей (всего 8 дублированных записей в таблице назначения в 11:00)

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

(Я не могу удалить промежуточную таблицу. Сейчас я фильтрую целевую таблицу, чтобы выбрать только уникальные записи. Есть ли лучший способсделать это?)

1 Ответ

0 голосов
/ 10 мая 2018

Ваши таблицы этапов и назначения находятся в Postgres, поэтому просто напишите логику, которая сравнивает данные в таблице этапов с таблицей dest и удаляет все записи из этапа, которые уже существуют в таблице dest.

DELETE FROM staging
WHERE EXISTS(SELECT 1 FROM dest WHERE dest.id = staging.id);

https://www.postgresql.org/docs/8.1/static/functions-subquery.html

...