Как я могу эффективно предотвратить дублирование строк в моей таблице фактов?

Question

Я построил конвейер фабрики данных, который ETL-данные из озера данных в хранилище данных. Я выбрал тип SCD 1 для своих размеров.

Мой конвейер содержит следующие действия:

1. [Хранимая процедура] Очистить промежуточные таблицы;
2. [Хранимая процедура] Получите метку времени последнего успешного обновления;
3. [U-SQL] Извлечь данные измерений из отфильтрованных файлов (тех, которые были изменены с момента последнего успешного обновления) в озере данных Azure, преобразовать их и вывести в виде CSV-файла;
4. [Копировать данные] Загрузить CSV в таблицу промежуточных измерений хранилища данных SQL;
5. [Хранимая процедура] Объединить данные из промежуточной таблицы в рабочую таблицу;
6. [U-SQL] Извлечь данные фактов из файлов (которые были изменены с момента последнего успешного обновления) в озере данных Azure, преобразовать их и вывести в виде файла CSV;
7. [Копировать данные] Загрузить CSV в таблицу фактов хранилища данных SQL;
8. [Хранимая процедура] Обновление метки времени успешного обновления.

Проблема с этим конвейером состоит в том, что я получаю дублированные записи фактов на своем складе, если запускаю конвейер дважды.

Вопрос

Как эффективно предотвратить дублирование строк в моей таблице фактов, учитывая все неподдерживаемые функции в хранилище данных SQL Azure?

Обновление

Я прочитал другую информацию об индексировании (и статистике) хранилища и о том, как его нужно восстанавливать после обновления.

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

Answer 1

Поднимитесь ли вы в хранилище данных SQL Azure ... ваша производительность значительно улучшится, и ваша проблема исчезнет.

Сколько строк в ваших отфильтрованных файлах?Если это от миллионов до десятков миллионов, я думаю, что вы, вероятно, можете избежать фильтра на этапе озера данных.Производительность Polybase + SQL должна превышать дополнительный объем данных.

Если вы можете избежать фильтра, используйте эту логику и отбросьте обработку U-SQL:

• Загрузка файлов впромежуточная таблица с подходящим распределением хеша
• Возьмите последнюю версию каждой строки (подходит для SCD1)
• Этап объединения с фактом с помощью запроса, подобного следующему:

BK= Бизнес-ключ столбец / ы.COLn = неключевые столбцы

-- Get latest row for each business key to eliminate duplicates.

create table stage2 with (heap,distribution = hash(bk)) as
select  bk,
        col1,
        col2,
        row_number() over (partition by bk order by timestamp desc) rownum
from    stage
where   rownum = 1;

-- Merge the stage into a copy of the dimension

create table dimension_copy with (heap,distribution=replicate) as

select    s.bk,
          s.col1,
          s.col2
from      stage2 s
where     not exists (
              select  1
              from    schema.dimension d
              where   d.bk = s.bk)

union

select   d.bk,
         case when s.bk is null then d.col1 else s.col1 end,
         case when s.bk is null then d.col2 else s.col2 end
from     dimension d
         left outer join stage2 s on s.bk = d.bk;

-- Switch the merged copy with the original 

alter table dimension_copy switch to dimension with (truncate_target=on);

-- Force distribution of replicated table across nodes

select top 1 * from dimension;