Блог

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

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

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

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

Вопрос

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

Обновить

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

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

Выполните удаление в хранилище данных 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;