cumulative table design

Дизайн накопительной таблицы

Проектирование сводных таблиц — чрезвычайно мощный инструмент обработки данных, который должен знать каждый инженер данных.

Эта конструкция создает таблицы, которые могут обеспечить эффективный анализ на произвольно больших (до тысяч дней) временных интервалах.

Вот схема высокоуровневого проекта конвейера для этого шаблона:

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

После получения ежедневных показателей мы FULL OUTER JOIN вчерашней сводной таблицы с сегодняшними ежедневными данными и строим массивы показателей для каждого пользователя. Это позволяет нам ввести новую историю без необходимости ее полного сканирования. (большой прирост производительности)

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

Чем дольше период вашего анализа, тем более важной становится эта закономерность!

Пример совокупной активности и вовлеченности пользователей

Весь синтаксис запросов использует синтаксис и функции Presto/Trino. Этот пример необходимо будет изменить для других вариантов SQL!

Мы будем использовать даты:

• 01.01.2022, поскольку сегодня в терминах Airflow это {{ ds }}
• 31 декабря 2021 г., поскольку вчера в терминах шаблонов Airflow это {{ yesterday_ds}}

В этом примере мы рассмотрим, как создать этот дизайн для подсчета ежедневных, еженедельных и ежемесячных активных пользователей, а также лайков, комментариев и репостов пользователей.

Наша исходная таблица в данном случае — event .

• Пользователь активен в любой день, если он генерирует событие для этого дня.
• В таблице есть event_type , like , comment , share или view

Заманчиво думать, что решение этой проблемы — запустить конвейер, что-то вроде

    SELECT 
        COUNT(DISTINCT user_id) as num_monthly_active_users,
        COUNT(CASE WHEN event_type = 'like' THEN 1 END) as num_likes_30d,
        COUNT(CASE WHEN event_type = 'comment' THEN 1 END) as num_comments_30d,
        COUNT(CASE WHEN event_type = 'share' THEN 1 END) as num_shares_30d,
        ...
    FROM events
    WHERE event_date BETWEEN DATE_SUB('2022-01-01', 30), AND '2022-01-01'

Проблема в том, что мы каждый день сканируем данные о событиях за 30 дней, чтобы получить эти цифры. Довольно расточительный, но простой конвейер. Должен быть способ, при котором нам нужно будет сканировать данные о событиях только один раз и объединять их с результатами за предыдущие 29 дней, верно? Можем ли мы создать структуру данных, в которой специалист по данным сможет запрашивать наши данные и легко узнавать количество действий, предпринятых пользователем за последние N дней?

Эта конструкция довольно проста и состоит всего из 3 шагов:

Шаг дневной таблицы

• На этом этапе мы агрегируем только события сегодняшнего дня, чтобы увидеть, кто активен ежедневно. Схема таблицы здесь
• Этот запрос довольно прост и понятен здесь
  • GROUP BY user_id , а затем считать их ежедневными активными, если у них есть какие-либо события.
  • мы добавляем несколько COUNT(CASE WHEN event_type = 'like' THEN 1 END) , чтобы подсчитать количество ежедневных лайков, комментариев и репостов.

Шаг накопления

• Схема таблицы для этого шага находится здесь.
• Запрос для этого шага намного сложнее. Это здесь
  • На этом этапе мы берем сегодняшние данные из дневной таблицы и вчерашние данные из кумулятивной таблицы.
  • Мы FULL OUTER JOIN этих двух наборов данных на today.user_id = yesterday.user_id
  • Если пользователь новый, его не будет во вчерашних данных. Кроме того, если пользователь сегодня не был активен, его нет в сегодняшних данных.
  • Поэтому нам нужно COALESCE(today.user_id, yesterday.user_id) as user_id , чтобы отслеживать всех пользователей.
  • Далее мы хотим построить столбец activity_array . Мы хотим, чтобы activity_array хранил данные только за последние 30 дней.
    • Поэтому мы проверяем, является ли CARDINALITY(activity_array) < 30 , чтобы понять, можем ли мы просто добавить сегодняшнее значение в начало массива или нам нужно отрезать элемент от конца массива, прежде чем добавлять сегодняшнее значение в начало массива.
    • Нам нужно выполнить COALESCE(t.is_active_today, 0) чтобы поместить нулевые значения в массив, когда пользователь неактивен.
  • Мы следуем шаблону, очень похожему на то, как мы создавали activity_array но также для лайков, комментариев и репостов!
  • После того, как мы создадим все наши массивы показателей, вычислить еженедельную и ежемесячную активность будет довольно просто.
    • CASE WHEN ARRAY_SUM(activity_array) > 0 THEN 1 ELSE 0 END дает нам ежемесячные активы, поскольку мы ограничиваем размер массива до 30
    • CASE WHEN ARRAY_SUM(SLICE(activity_array, 1, 7)) > 0 THEN 1 ELSE 0 END дает нам еженедельную активность, поскольку мы проверяем только первые 7 элементов массива (т.е. последние 7 дней)
  • Подсчитать количество лайков, комментариев и репостов за последние 7 дней тоже легко.
    • ARRAY_SUM(SLICE(like_array, 1, 7)) as num_likes_7d дает нам количество лайков, которые этот пользователь поставил за последние 7 дней.
    • ARRAY_SUM(like_array) as num_likes_30d дает нам количество лайков, которые этот пользователь поставил за последние 30 дней, поскольку размер массива фиксирован.

Шаг DAG

• Пример DAG можно найти здесь.
• Ключевые вещи, которые следует помнить о DAG:
  • Совокупные группы обеспечения доступности баз данных всегда должны иметь depends_on_past: True
  • Шаг ежедневного агрегирования должен быть выше этапа кумуляции.