ETL в Databricks SQL

При работе с большими объемами данных требуется конвейер, который может обрабатывать только новые и измененные записи вместо повторной обработки всего набора данных. Это называется добавочным ETL. В Databricks SQL можно создавать добавочные конвейеры ETL с помощью потоковых таблиц и материализованных представлений без написания процедурного кода или планирования обновлений вручную.

В этом руководстве описан общий шаблон: отслеживание изменений продукта с течением времени. Вы создаёте исходную таблицу, фиксируете события изменений, строите таблицу измерений, которая сохраняет полную историю каждого продукта, ведёте таблицу фактов, которая постепенно перезаписывает недавние записи, и добавляете поверх агрегатный слой отчетности. Вы используете AUTO CDC для таблицы измерений и поток REPLACE WHERE для таблицы фактов.

Ключевым компонентом этого руководства является AUTO CDC. В традиционном хранилище вам пришлось бы писать сложные MERGE INTO инструкции для согласования событий вставки, обновления и удаления в целевую таблицу. Этот подход подвержен ошибкам, особенно когда события приходят в неправильном порядке. AUTO CDC обрабатывает это для вас. Вы указываете бизнес-ключ, столбец последовательности, а также выбираете, хотите ли использовать SCD Type 1 (только последнее значение) или SCD Type 2 (полная история), и Azure Databricks автоматически применяет правильную логику слияния. Общие сведения о CDC см. в api-интерфейсах AUTO CDC: упрощение отслеживания измененных данных с помощью конвейеров.

К концу этого руководства вы будете иметь следующее:

  1. Создана исходная таблица, которая отслеживает изменения с помощью потока данных об изменениях.
  2. Проверьте необработанные данные об изменениях, чтобы понять поток событий CDC.
  3. Используется AUTO CDC для создания таблицы измерений SCD Типа 2 из этих событий.
  4. События удаления обрабатываются поэтапно через конвейер.
  5. Построил таблицу фактов, которая инкрементально перезаписывает недавние строки с потоком REPLACE WHERE.
  6. Создано материализованное представление, которое пошагово обновляет сводный отчет.
  7. Настроено SCHEDULE REFRESH EVERY 1 DAY , чтобы изменения распространялись автоматически через конвейер.

Требования

Чтобы завершить работу с этим руководством, необходимо выполнить следующие требования:

Шаг 1. Настройка каталога и схемы

Откройте редактор SQL Databricks и задайте рабочий каталог и схему. У вас должно быть разрешение на USE выбор каталога и схемы:

USE CATALOG <your-catalog>;
USE SCHEMA <your-schema>;

Шаг 2. Создание исходной таблицы и загрузка данных

Создайте productsтаблицу с включенной функцией Use change data feed on Azure Databricks (CDF). CDF — это функция Delta Lake, которая записывает каждую вставку, обновление и удаление в виде запрашиваемого журнала изменений. Это похоже на поток CDC из исходной системы транзакций, за исключением того, что изменения фиксируются непосредственно в таблице Delta, а не из внешнего журнала. Здесь вы используете CDF для создания событий изменений, которые будет использоваться подчиненным конвейером.

  1. Создайте таблицу и загрузите начальные записи:

    CREATE OR REPLACE TABLE products (
  product_id INT,
  product_name STRING,
  category STRING,
  warehouse STRING
)
TBLPROPERTIES (delta.enableChangeDataFeed = true);

INSERT INTO products VALUES
  (1, 'Spoon', 'Cutlery', 'Seattle'),
  (2, 'Fork', 'Cutlery', 'Portland'),
  (3, 'Knife', 'Cutlery', 'Denver'),
  (4, 'Chair', 'Furniture', 'Austin'),
  (5, 'Table', 'Furniture', 'Chicago'),
  (6, 'Lamp', 'Lighting', 'Boston'),
  (7, 'Mug', 'Kitchenware', 'Seattle'),
  (8, 'Plate', 'Kitchenware', 'Atlanta'),
  (9, 'Bowl', 'Kitchenware', 'Dallas'),
  (10, 'Glass', 'Kitchenware', 'Phoenix');

  2. Имитируйте изменения вышестоящего потока, включая новые продукты, перемещение склада и переназначение категории:

    INSERT INTO products VALUES
  (11, 'Napkin', 'Dining', 'San Francisco'),
  (12, 'Coaster', 'Dining', 'New York');

UPDATE products SET warehouse = 'Los Angeles' WHERE product_id = 1;
UPDATE products SET category = 'Dining' WHERE product_id = 2;

Шаг 3. Запрос потока измененных данных

Прежде чем создавать низовой конвейер, полезно просмотреть события сырого изменения, чтобы понять, что AUTO CDC будет обрабатываться. Функция table_changes() считывает журнал CDF и возвращает каждую записанную операцию вместе со столбцами метаданных:

SELECT
  product_id, product_name, warehouse,
  _change_type, _commit_version
FROM table_changes('products', 1)
ORDER BY _commit_version, product_id;

Например, у Ложки есть три события: одно insert (Сиэтл), другое update_preimage (Сиэтл) и еще одно update_postimage (Лос-Анджелес).

Обратите внимание, что одно логическое изменение (например, перемещение ложки в другое хранилище) создает несколько событий: предварительное создание и запись. В традиционном хранилище вы напишете MERGE инструкцию, чтобы примирить все эти события в целевую таблицу, обрабатывать вставки, обновления и удаления с отдельной логикой и убедиться, что события применяются в правильном порядке. Это именно сложность, которая AUTO CDC устраняется на следующем шаге.

Шаг 4. Создание измерения SCD Типа 2 с помощью AUTO CDC

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

AUTO CDC добавляет обработку отслеживания измененных данных поверх потоковой таблицы. Вместо написания инструкции MERGE INTO, которая вручную обрабатывает вставки, обновления и удаления, вы объявляете бизнес-ключ и последовательный столбец и позволяет Azure Databricks применить правильную логику. AUTO CDC также обрабатывает события вне порядка автоматически, что является распространенной проблемой при обработке MERGE INTO событий, поступающих из распределенных систем или пакетной загрузки с перекрывающимися метками времени.

Следующая инструкция создает таблицу SCD Type 2, которая сохраняет полную историю версий каждого продукта. Каждая версия получает __START_AT и __END_AT метки времени. NULL в __END_AT помечает текущую версию.

CREATE OR REFRESH STREAMING TABLE products_history
SCHEDULE REFRESH EVERY 1 DAY
FLOW AUTO CDC
FROM (
  SELECT * FROM STREAM products WITH (readChangeFeed = true)
  WHERE _change_type != 'update_preimage'
)
KEYS (product_id)
APPLY AS DELETE WHEN _change_type = 'delete'
SEQUENCE BY _commit_timestamp
COLUMNS * EXCEPT (_change_type, _commit_version, _commit_timestamp)
STORED AS SCD TYPE 2;
  • SCHEDULE REFRESH EVERY 1 DAY: обновляет таблицу по ежедневному расписанию.
  • FLOW AUTO CDC: обозначает это как поток CDC. Azure Databricks автоматически применяет семантику вставки, обновления и удаления.
  • WHERE _change_type != 'update_preimage': пропускает предобразы. В postimage уже есть обновлённая строка.
  • KEYS (product_id): бизнес-ключ. События с тем же ключом объединяются в версионные строки.
  • APPLY AS DELETE WHEN _change_type = 'delete': закрывает текущую версию при поступлении события удаления. Это позволяет определить условие, определяющее событие удаления.
  • SEQUENCE BY _commit_timestamp: устанавливает порядок событий. Правильно обрабатывает поступления вне порядка.
  • STORED AS SCD TYPE 2: сохраняет полную историю. AUTO CDC поддерживает как SCD Type 1, так и SCD Type 2.

Запросите таблицу измерений:

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;
  • Ложка: две версии. Сиэтл (закрыт, набор __END_AT) и Лос-Анджелес (действующий, __END_AT = NULL).
  • Вилка: две версии. Категория столовых приборов (закрытая) и категория "Столовая" (текущая).
  • Салфетка и Подставка: по одной версии каждой (только что вставленные, __END_AT = NULL).
  • Все остальные продукты: одна версия каждой (__END_AT = NULL).

Шаг 5. Процесс удаления через конвейер

Теперь имитируйте два прекращенных продукта, удалив их из исходной таблицы:

DELETE FROM products WHERE product_id = 9;
DELETE FROM products WHERE product_id = 10;

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

REFRESH STREAMING TABLE products_history;

Запросите таблицу измерений, чтобы убедиться, что были применены удаления:

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;

Чаша и бокал теперь считаются закрытыми с комплектом __END_AT, отметив их как снятые с производства. Все остальные текущие продукты остаются неизменными. Таблица потоковой передачи обрабатывает только новые события удаления без повторной обработки вставок и обновлений из предыдущего обновления.

Шаг 6: Постройте таблицу фактов с потоком REPLACE WHERE

Таблица измерений в предыдущих шагах использует AUTO CDC Slowly Changing Dimension, потому что отслеживает текущее и историческое состояние каждого продукта. К таблицам фактов предъявляются разные требования. Таблица фактов фиксирует события, такие как продажи, и обычно вы повторно обрабатываете только недавний диапазон данных, например, для исправления записей, поступивших с задержкой. Для этого шаблона используйте инкрементальный поток REPLACE WHERE, который перезаписывает только строки, соответствующие предикату, и оставляет все остальные строки нетронутыми. Для полного сравнения этой конструкции с материализированными видами см. раздел «Когда использовать REPLACE WHERE вместо материализированных видов».

  1. Создайте product_sales таблицу и загрузите несколько событий с продажами, некоторые из которых были недавними, а некоторые старые:

    CREATE OR REPLACE TABLE product_sales (
  sale_id INT,
  product_id INT,
  quantity INT,
  sale_date DATE
);

INSERT INTO product_sales VALUES
  (101, 1, 3, current_date()),
  (102, 2, 1, current_date()),
  (103, 4, 5, date_add(current_date(), -2)),
  (104, 6, 2, date_add(current_date(), -10));

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

    CREATE OR REFRESH STREAMING TABLE product_sales_recent
SCHEDULE REFRESH EVERY 1 DAY
FLOW REPLACE WHERE sale_date >= date_add(current_date(), -7) BY NAME
SELECT sale_id, product_id, quantity, sale_date
FROM product_sales;
    • FLOW REPLACE WHERE sale_date >= date_add(current_date(), -7): при каждом обновлении удаляет строки в целевой таблице, соответствующие предикату, пересчитывает исходный запрос для того же диапазона и вставляет результаты. Строки за пределами 7-дневного окна остаются нетронутыми.
    • SCHEDULE REFRESH EVERY 1 DAY: обновляет таблицу по тому же ежедневному расписанию, что и таблица измерений.
    • BY NAME: требуется. Сопоставляет столбцы по названию, а не по позиции.

    Запись о продаже, совершённая 10 дней назад, выходит за рамки предиката, поэтому поток не загружает её. Сохраняются только продажи за последние 7 дней.

  3. Запросите фактовую таблицу:

    SELECT sale_id, product_id, quantity, sale_date
FROM product_sales_recent
ORDER BY sale_date DESC, sale_id;

    Результат содержит три продажи за последние 7 дней. Значения sale_date зависят от дня, в который вы запускаете руководство:

    sale_id product_id количество дата_продажи
    101 1 3 Сегодня
    102 2 1 Сегодня
    103 4 5 2 дн. назад

    Продажа 104 не включена в результат, потому что её дата находится вне 7-дневного окна.

Шаг 7: Создайте агрегированное материализованное представление

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

Материализованное представление сохраняет предварительно вычисляемые результаты запроса в виде физической таблицы. В отличие от обычного представления, которое повторно выполняет запрос при каждом чтении из него, материализованное представление сохраняет результаты и пересчитывает только строки, затронутые изменениями в исходных данных, при каждом обновлении. Он хорошо подходит для панелей мониторинга и отчетов, где важна производительность запросов.

CREATE OR REPLACE MATERIALIZED VIEW products_by_category
SCHEDULE REFRESH EVERY 1 DAY
AS
SELECT
  category,
  COUNT(*) AS active_products
FROM products_history
WHERE __END_AT IS NULL
GROUP BY category;

SCHEDULE REFRESH EVERY 1 DAY означает, что это представление обновляется по ежедневному расписанию. В сочетании с тем же расписанием в таблице потоковой передачи теперь есть трехэтапный конвейер, в котором изменения в исходной таблице каскадируются через измерение и агрегируются в каждом цикле обновления. Вам не нужно вручную обновлять измерение или агрегат.

SELECT * FROM products_by_category ORDER BY active_products DESC;

Шаг 8: Проверьте каскад целиком

Чтобы проверить полный каскад обработки, внесите изменение в исходную таблицу и зарегистрируйте продажу, поступившую с опозданием, за вчера:

UPDATE products SET warehouse = 'Seattle' WHERE product_id = 3;

INSERT INTO product_sales VALUES (105, 3, 4, date_add(current_date(), -1));

Нож перемещается из Денвера в Сиэтл. UPDATE запускает полный каскад конвейера, демонстрируя, как три этапа работают вместе:

  1. products фиксирует событие изменения в ленте данных изменений.
  2. products_history обрабатывает событие и добавляет новую версию для ножа.
  3. products_by_category пересчитывает только затронутую строку столовых приборов.

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

REFRESH STREAMING TABLE product_sales_recent;

Поток пересчитывает последние 7 дней из product_sales, так что продажа 105 теперь появляется в product_sales_recent.

Проверить.

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
WHERE product_id = 3
ORDER BY __START_AT;

SELECT * FROM products_by_category ORDER BY active_products DESC;

SELECT sale_id, product_id, quantity, sale_date
FROM product_sales_recent
ORDER BY sale_date DESC, sale_id;

Очистка

Чтобы очистить ресурсы, созданные этим руководством, используйте следующий SQL:

DROP MATERIALIZED VIEW IF EXISTS products_by_category;
DROP TABLE IF EXISTS product_sales_recent;
DROP TABLE IF EXISTS product_sales;
DROP TABLE IF EXISTS products_history;
DROP TABLE IF EXISTS products;

Дополнительные ресурсы

  • Последнее обновление: