Решение проблемы сложности объединения снимков

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

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

Какая ваша первая мысль? Ну, вы опытный инженер по аналитике и знаете, что хорошие люди из dbt Labs имеют решение для вас. И тут вас осеняет — ответ в снимках!

Что такое снимки и где они становятся сложными?

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

Вот пример набора данных.

entity_id	important_status	updated_at
1	available	2021-11-15 16:00:0000
2	not_available	2021-11-15 15:30:0000

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

entity_id	important_status	dbt_valid_from	dbt_valid_to
1	available	2021-11-15 16:00:0000	NULL
1	pending	2021-11-10 08:00:000	2021-11-15 16:00:0000
1	not_available	2021-10-01 10:00:000	2021-11-10 08:00:000
2	not_available	2021-11-15 15:30:0000	NULL

Снимки невероятно полезны, но они добавляют немного сложности при объединении таблиц вниз по потоку, потому что вы добавили несколько строк истории на каждый id. Что происходит, когда у вас есть 10 снимков, которые вы хотите объединить, и вы хотите зафиксировать историю всех наборов данных?

Рассмотрим сложность проблемы: вы успешно зафиксировали историю всех ваших таблиц, используя снимки. У вас есть history_table_1 и history_table_2, и вы хотите объединить их по общему ключу, product_id. Однако мы не можем просто объединить по первичному ключу, потому что каждая таблица имеет несколько строк истории для одного и того же id, все действительные в разные временные промежутки.

history_table_1:

product_id	important_status	dbt_valid_from	dbt_valid_to
1	available	2021-11-15 16:00:0000	NULL
1	pending	2021-11-10 08:00:000	2021-11-15 16:00:0000
1	not_available	2021-10-01 10:00:000	2021-11-10 08:00:000
2	not_available	2021-11-15 15:30:0000	NULL

history_table_2:

product_id	order_id	product_order_id	order_status	dbt_valid_from	dbt_valid_to
1	A	1A	available	2021-11-15 16:00:0000	NULL
1	A	1A	pending	2021-10-31 12:00:000	2021-11-15 16:00:0000
1	B	1B	available	2021-11-15 15:30:0000	NULL
1	B	1B	pending	2021-11-10 10:00:000	2021-11-15 15:30:0000
2	C	2C	available	2021-11-10 15:00:0000	NULL

Это не выглядит так уж плохо. Насколько это может быть сложно? Давайте посмотрим на математику. Скажем, historical_table_1 имеет x исторических строк на product_id, и y id всего. Это x*y = n строк данных. historical_table_2 имеет z исторических строк на product_id, и w id (z*w = m строк). Последующее объединение по product_id затем изменяет сложность с O(n) на O(n*m) очень быстро (x*y*z*w возможностей!). Сложность продолжает увеличиваться по мере объединения все большего количества исторических таблиц.

Я знаю, о чем вы думаете — какой беспорядок! Не можем ли мы просто все объединить и сделать снимок полученной таблицы? Это не такая уж плохая мысль. Это избавило бы вас от необходимости продумывать проблему с O(n*m*a*b*c*d*...*q) сложностью. И в некоторых случаях это может зафиксировать всю историю, которая вам нужна!

Однако это не предоставляет решения проблемы, поставленной изначально. Исторические записи отслеживают, когда каждая таблица действительна, а не когда действительна объединенная таблица, и эта история для каждого набора данных будет отражена только тогда, когда вы сделаете снимок каждой таблицы, а затем объедините их, а не объедините и затем сделаете снимок таблицы. valid_from и valid_to, встроенные в объединенную-затем-снимочную таблицу, будут построены только из временных меток updated_at, где обновляется объединенная таблица, и, следовательно, изменения в исходных данных могут не быть зафиксированы. Мы хотим понять, когда записи действительно действительны во всех таблицах, что означает, что нам нужно учитывать действительные временные метки из каждого отдельного набора данных.

Хорошо, мы исключили легкий способ решения этого вопроса. Так что давайте решим эту проблему O(n*m*a*b*c*d*...*q)! Мы можем это сделать.

План действий для нашего решения

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

Для historical_table_1 и historical_table_2 мы будем объединять по product_id, где historical_table_1.valid_from до historical_table_1.valid_to имеет пересекающееся время с historical_table_2.valid_from до historical_table_2.valid_to.

Это сводится к следующим шагам:

1. Избавьтесь от дубликатов, если это необходимо
2. Сделайте снимок ваших таблиц данных
3. Защитите ваши даты valid_to на будущее
4. Объедините ваши таблицы с несовпадающим зерном, чтобы построить разветвленный стержень, содержащий зерновые id, к которым мы будем присоединять остальные данные
5. Присоедините снимки к стержню данных по соответствующему id в пересекающихся временных промежутках, сужая действительные временные промежутки на строку по мере присоединения большего количества таблиц
6. Очистите ваши столбцы в финальной
7. Необязательное добавление глобальной переменной для фильтрации только текущих значений

Итак, давайте погрузимся! С головой! Шаг 1 описан в этом блоге: Как удалить частичные дубликаты. Его нужно реализовать только в том случае, если вы имеете дело с дубликатами в ваших данных. Если у вас нет дубликатов в данных (вау! пришлите мне номер вашего инженера по данным СРОЧНО), вы можете пропустить этот шаг.

Шаг 1: Убедитесь, что ваши таблицы данных не содержат дубликатов

Что происходит на этом шаге?

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

Почему этот шаг важен? Потому что вы будете объединять так много строк по одному и тому же id, и действительные временные метки для каждой строки будут определять точное место для объединения одной таблицы с другой. Мы не можем сделать это точно с дубликатами! (Но также, вы должны проверять на наличие дубликатов в любом случае, потому что мы инженеры по аналитике, верно?)

Смотрите этот блог для удаления частичных дубликатов!

Шаг 2: Сделайте снимок ваших данных

Что происходит на этом шаге?

Шаг 2 проведет вас через процесс создания снимка ваших данных. Пример предполагает, что вы прошли Шаг 1, но если вы пропустили этот шаг, просто сделайте снимок ваших данных, основываясь на ссылках, предоставленных ниже.

Знаете ли вы, как сделать снимок данных? Это простой блок Jinja с некоторыми указанными конфигурациями. Существует так много объяснений, как реализовать это, поэтому я не буду вас утомлять. Но вы знаете, что я дам вам несколько ссылок. Бум. И фубар!

Вы можете сделать снимок, проверяя ваш change_id, если вы реализовали логику удаления дубликатов из Шага 1, или используя стратегию временных меток, если у вас есть надежная временная метка.

{% snapshot snp_product %}
{{
   config(
     target_schema=generate_schema_name('snapshots'),
     unique_key='assetid',
     strategy='check',
     check_cols=['change_id']
   )
}}
select * from {{ ref('base_product') }}
{% endsnapshot %}

Шаг 3: Защитите ваши временные метки на будущее

Что происходит на этом шаге?

Шаг 3 проведет вас через процесс замены значения valid_to = NULL в вашем снимке на дату, защищенную на будущее, чтобы обеспечить плавное объединение снимков.

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

Примечание

Это отличное место для установки глобальной переменной! Вы можете определить вашу переменную, защищенную на будущее, в файле dbt_project.yml.

vars:
 future_proof_date: '9999-12-31'

И используйте coalesce!

coalesce(dbt_valid_to, cast('{{ var("future_proof_date") }}' as timestamp)) as valid_to

Вы поблагодарите себя позже за создание глобальной переменной. Добавление важных глобальных переменных обеспечит успех вашего будущего "я". Теперь вы можете фильтровать все ваши данные до текущего состояния, просто фильтруя по where valid_to = future_proof_date. Вы также можете убедиться, что все "медведи данных" с их "лапами данных" в "банке данных" ссылаются на одну и ту же future_proof_date, а не на 9998-12-31, или 9999-12-31, или 10000-01-01, что в конечном итоге что-то сломает. Вы знаете, что это произойдет; не спорьте со мной! Глобальные переменные для победы!

Шаг 4: Объедините ваши таблицы, чтобы построить разветвленный стержень id

Что происходит на этом шаге?

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

Давайте посмотрим, как мы это сделаем на примере. У вас может быть много событий, связанных с одним product_id. Каждый product_id может иметь несколько order_ids, и каждый order_id может иметь другой id, связанный с ним. Это означает, что зерно каждой таблицы должно быть идентифицировано. Суть здесь в том, что нам нужно построить id на самом тонком зерне. Для этого мы добавим dbt_utils.generate_surrogate_key в модели стадирования, которые находятся поверх таблиц снимков.

Затем, в вашей модели объединения, давайте добавим CTE, чтобы построить наш стержень с нашими id этих разных зерен.

build_spine as (
	historical_table_1.*,
	historical_table_2.product_order_id,
	historical_table_3.other_entity_grain_id, 
	...
from historical_table_1
left join
	historical_table_2 
	on historical_table_1.product_id = historical_table_2.product_id
left join 
	historical_table_3
	on historical_table_1.product_id = historical_table_3.product_id
... )

Результатом будут все столбцы из вашей первой таблицы, разветвленные как можно больше за счет добавленных столбцов id. Мы будем использовать эти столбцы id, чтобы присоединить исторические данные из наших таблиц. Крайне важно отметить, что если у вас есть таблицы, которые являются частью этого шаблона и захватываются на том же зерне, что и исходная таблица, вы не хотите присоединять эту таблицу и id как часть стержня. Это разветвится слишком сильно и вызовет дубликаты в ваших данных. Вместо этого просто присоедините таблицы с тем же зерном, что и исходная таблица (в данном случае, historical_table_1 по product_id) на следующем шаге, используя макрос.

product_id	important_status	dbt_valid_from	dbt_valid_to	product_order_id
1	available	2021-11-15 16:00:0000	NULL	1A
1	pending	2021-11-10 08:00:000	2021-11-15 16:00:0000	1A
1	not_available	2021-10-01 10:00:000	2021-11-10 08:00:000	1A
1	available	2021-11-15 16:00:0000	NULL	1B
1	pending	2021-11-10 08:00:000	2021-11-15 16:00:0000	1B
1	not_available	2021-10-01 10:00:000	2021-11-10 08:00:000	1B
2	not_available	2021-11-15 15:30:0000	NULL	2C

Шаг 5: Присоедините ваши снимки к стержню id

Что происходит на этом шаге?

Шаг 5 проведет вас через логику объединения снимков и макрос, который упростит объединения.

Теперь я рекомендую вам строить отдельные CTE с одним объединением за раз. Почему мы строим CTE с одним объединением, а не все объединения в одном? Так много причин, но есть две большие.

• Во-первых, это сложно. Вам нужно будет устранять неполадки, и самый простой способ сделать это — разделить логику объединения на отдельные CTE. Построив ваш код таким образом, вы можете легко добавить select * from last_cte, чтобы проверить, что ваша логика делает то, что вы думаете, прежде чем добавлять более сложные объединения.
• Во-вторых, вы используете значения valid_from и valid_to каждой вновь присоединенной таблицы, чтобы определить новые valid_from и valid_to временные метки для результирующей таблицы — где обе строки действительны в одни и те же временные промежутки. Хотя вы могли бы сделать это в одном большом массивном объединении, это станет очень сложным и трудным для устранения неполадок, когда вы столкнетесь с странными результатами.

DRY — время макросов!

Этот макрос завершает ваш CTE объединения, что позволяет вам добавлять столбцы из новой таблицы, которую вы присоединяете, перед вызовом макроса. Он также предполагает, что вы заменили ваши valid_to = NULL на фактический тип даты с фактической датой, указывающей, что строка в настоящее время действительна.

Ваши параметры: cte_join, таблица, создающая стержень вашей финальной модели, cte_join_on, которая является новой таблицей, которую вы присоединяете к стержню. Значения valid_to и valid_from для обеих этих таблиц, и id, по которым вы присоединяетесь (названы дважды на случай, если у них разные имена столбцов, но в большинстве случаев эти два параметра будут иметь одно и то же имя).

-- требует, чтобы любые дополнительные столбцы из table_join_on были перечислены до использования этого макроса.
-- предполагает, что мы заменили случаи valid_to = null на future_proof_date = '9999-12-31'.
 
{% macro join_snapshots(cte_join, cte_join_on, cte_join_valid_to,
   cte_join_valid_from, cte_join_on_valid_to, cte_join_on_valid_from,
   cte_join_id, cte_join_on_id) %}
 
 
       {{cte_join}}.*,
       greatest({{cte_join}}.{{cte_join_valid_from}},
               coalesce( {{cte_join_on}}.{{cte_join_on_valid_from}}, {{cte_join}}.{{cte_join_valid_from}}))
           as add_{{cte_join_on}}_valid_from,
       least({{cte_join}}.{{cte_join_valid_to}},
           coalesce({{cte_join_on}}.{{cte_join_on_valid_to}}, {{cte_join}}.{{cte_join_valid_to}})) as add_{{cte_join_on}}_valid_to
  
   from {{cte_join}}
   left join {{cte_join_on}} on {{cte_join}}.{{cte_join_id}} = {{cte_join_on}}.{{cte_join_on_id}}
      and ({{cte_join_on}}.{{cte_join_on_valid_from}} <= {{cte_join}}.{{cte_join_valid_to}}
      and {{cte_join_on}}.{{cte_join_on_valid_to}} >= {{cte_join}}.{{cte_join_valid_from}})
      
  
{% endmacro %}

Логика объединения находит, где id совпадают и где временные метки пересекаются между двумя таблицами. Мы используем наибольший valid_from и наименьший valid_to между двумя таблицами, чтобы гарантировать, что новый, суженный временной промежуток для строки — это когда строки из обеих таблиц действительны. Обновление: Особая благодарность Аллину Опицу за упрощение этой логики объединения! Теперь она выглядит намного красивее.

Вы должны увидеть что-то вроде этого в качестве конечного результата:

product_id	product_order_id	order_id	important_status	order_status	greatest_valid_from	least_valid_to
1	1A	A	available	available	2021-11-15 16:00:0000	9999-12-31
1	1A	A	pending	pending	2021-11-10 08:00:000	2021-11-15 16:00:0000
1	1A	A	not_available	pending	2021-10-31 10:00:000	2021-11-10 08:00:000
1	1B	B	available	available	2021-11-15 16:00:0000	9999-12-31
1	1B	B	pending	pending	2021-11-10 08:00:000	2021-11-15 16:00:0000
1	1B	B	not_available	pending	2021-10-01 10:00:000	2021-11-10 08:00:000
2	2C	C	not_available	available	2021-11-15 15:30:0000	9999-12-31

Продолжайте присоединять ваши таблицы и сужать ваши действительные временные промежутки.

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

Шаг 6: Очистите вашу финальную таблицу с помощью CTE (конечно!)

Что происходит на этом шаге?

Шаг 6 — это завершение вашего кода с финальным, очищенным CTE.

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

Шаг 7: Необязательно — добавьте глобальную переменную для построения исторических данных против текущих

Что происходит на этом шаге?

Шаг 7 проведет вас через возможность добавления глобальной переменной для запуска только самых актуальных данных.

Может быть полезно добавить переменную только для текущих записей, чтобы запустить ваш проект. Это быстрый способ пропустить исторические данные, не создавая новые модели или фильтруя вашу историческую таблицу. Вы можете настроить отдельную задачу для целевой новой схемы и построить таблицы только с текущими данными, которые готовы для отчетов на сегодняшний день. Вы поймете, что это подходит для вашего проекта, если у вас есть BI-инструмент, который не любит фильтровать большие таблицы с историей (например, Tableau), но предпочитает иметь легко доступные, готовые к запуску таблицы. Чтобы встроить эту функцию, добавьте глобальную переменную в dbt_project.yml, чтобы у вашей future_proof_date был друг:

future_proof_date: '9999-12-31'
current_records_only: true

И добавьте Jinja-if в ваши модели стадирования, чтобы ваш проект запрашивал только построение данных, которые актуальны, без необходимости переопределять ваши снимки:

{% if var("current_records_only") %}

where valid_to = cast('{{ var("future_proof_date") }}' as timestamp)

{% endif %}

Заключительные мысли

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