Как выделить пользовательские сессии с таймаутом 5 минут?

Как выделить пользовательские сессии с таймаутом 5 минут

Сессионирование (sessionization) — это важная техника в аналитике, которая группирует события пользователя в логические сессии на основе временных промежутков. Если между двумя событиями одного пользователя прошло более 5 минут, они считаются разными сессиями.

Основные подходы

1. SQL решение с оконными функциями (самый распространённый)

Это решение работает в Spark SQL, PostgreSQL, Hive, Clickhouse и других SQL движках:

WITH ordered_events AS (
    -- Сортируем события каждого пользователя по времени
    SELECT 
        user_id,
        event_timestamp,
        LAG(event_timestamp) OVER (PARTITION BY user_id ORDER BY event_timestamp) AS prev_event_time
    FROM events
)
,
session_markers AS (
    -- Вычисляем разницу со предыдущим событием
    SELECT 
        user_id,
        event_timestamp,
        prev_event_time,
        -- Если разница больше 5 минут (300 секунд), это начало новой сессии
        CASE 
            WHEN prev_event_time IS NULL THEN 1
            WHEN (unix_timestamp(event_timestamp) - unix_timestamp(prev_event_time)) > 300 THEN 1
            ELSE 0
        END AS is_new_session
    FROM ordered_events
)
,
session_groups AS (
    -- Суммируем флаги, чтобы получить ID сессии
    SELECT 
        user_id,
        event_timestamp,
        is_new_session,
        SUM(is_new_session) OVER (PARTITION BY user_id ORDER BY event_timestamp) AS session_id
    FROM session_markers
)

SELECT 
    user_id,
    CONCAT(user_id, '_', session_id) AS session_uid,
    MIN(event_timestamp) AS session_start,
    MAX(event_timestamp) AS session_end,
    COUNT(*) AS events_in_session,
    DATEDIFF(SECOND, MIN(event_timestamp), MAX(event_timestamp)) / 60.0 AS session_duration_minutes
FROM session_groups
GROUP BY user_id, session_id
ORDER BY user_id, session_start;

2. PostgreSQL с временными интервалами

WITH events_with_gaps AS (
    SELECT 
        user_id,
        event_timestamp,
        event_type,
        -- Флаг: если разница со предыдущим событием > 5 минут
        CASE 
            WHEN LAG(event_timestamp) OVER (PARTITION BY user_id ORDER BY event_timestamp) IS NULL 
            THEN TRUE
            WHEN EXTRACT(EPOCH FROM (event_timestamp - LAG(event_timestamp) OVER (PARTITION BY user_id ORDER BY event_timestamp))) > 300
            THEN TRUE
            ELSE FALSE
        END AS gap_detected
    FROM events
),
session_ids AS (
    SELECT 
        user_id,
        event_timestamp,
        event_type,
        gap_detected,
        -- Генерируем ID сессии
        ROW_NUMBER() OVER (PARTITION BY user_id) - 
        SUM(CASE WHEN gap_detected THEN 1 ELSE 0 END) 
            OVER (PARTITION BY user_id ORDER BY event_timestamp) AS session_number
    FROM events_with_gaps
)

SELECT 
    user_id,
    session_number,
    COUNT(*) AS events,
    MIN(event_timestamp) AS session_start,
    MAX(event_timestamp) AS session_end,
    MAX(event_timestamp) - MIN(event_timestamp) AS duration,
    STRING_AGG(event_type, ', ' ORDER BY event_timestamp) AS events_sequence
FROM session_ids
GROUP BY user_id, session_number
ORDER BY user_id, session_start;

3. Spark Python решение

from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import \
    col, lag, unix_timestamp, when, sum as spark_sum, \
    row_number, concat_ws, collect_list, first, last, count, min as spark_min, max as spark_max

spark = SparkSession.builder.getOrCreate()

# Предположим, у нас есть DataFrame events с колонками:
# user_id, event_timestamp, event_type

window = Window.partitionBy('user_id').orderBy('event_timestamp')

# Шаг 1: определяем границы сессий
events_with_gaps = events.withColumn(
    'prev_timestamp', lag('event_timestamp').over(window)
).withColumn(
    'gap_seconds', 
    when(
        col('prev_timestamp').isNotNull(),
        (unix_timestamp('event_timestamp') - unix_timestamp('prev_timestamp'))
    ).otherwise(0)
).withColumn(
    'is_new_session',
    when(col('gap_seconds') > 300, 1).otherwise(0)
)

# Шаг 2: генерируем ID сессии
session_ids = events_with_gaps.withColumn(
    'session_id',
    spark_sum('is_new_session').over(window)
)

# Шаг 3: агрегируем по сессиям
sessions = session_ids.groupBy('user_id', 'session_id').agg(
    spark_min('event_timestamp').alias('session_start'),
    spark_max('event_timestamp').alias('session_end'),
    count('*').alias('event_count'),
    collect_list('event_type').alias('events')
).orderBy('user_id', 'session_start')

sessions.show()

4. Hive SQL вариант

WITH raw_with_lags AS (
    SELECT 
        user_id,
        event_timestamp,
        event_type,
        LAG(event_timestamp, 1) OVER (PARTITION BY user_id ORDER BY event_timestamp) as prev_event_ts,
        (unix_timestamp(event_timestamp) - unix_timestamp(
            LAG(event_timestamp, 1) OVER (PARTITION BY user_id ORDER BY event_timestamp)
        )) / 60 as minutes_since_prev -- разница в минутах
    FROM events
    WHERE event_timestamp >= '2024-08-01' AND event_timestamp < '2024-09-01'
),
session_starts AS (
    SELECT 
        user_id,
        event_timestamp,
        event_type,
        CASE 
            WHEN prev_event_ts IS NULL THEN 1 -- первое событие
            WHEN minutes_since_prev >= 5 THEN 1 -- есть пауза >= 5 минут
            ELSE 0 
        END as session_start_flag
    FROM raw_with_lags
),
session_ids AS (
    SELECT 
        user_id,
        event_timestamp,
        event_type,
        SUM(session_start_flag) OVER (PARTITION BY user_id ORDER BY event_timestamp) as session_id
    FROM session_starts
)

SELECT 
    user_id,
    session_id,
    MIN(event_timestamp) as session_start,
    MAX(event_timestamp) as session_end,
    COUNT(*) as event_count,
    DATEDIFF(MAX(event_timestamp), MIN(event_timestamp)) as session_duration_minutes,
    COLLECT_LIST(event_type) as event_sequence
FROM session_ids
GROUP BY user_id, session_id
ORDER BY user_id, session_start;

Пример данных и результатов

Входные данные (events):

user_id | event_timestamp      | event_type
--------|----------------------|----------
  101   | 2024-08-30 10:00:00 | page_view
  101   | 2024-08-30 10:02:15 | click
  101   | 2024-08-30 10:04:30 | click
  101   | 2024-08-30 10:12:00 | page_view  -- 7.5 минут со предыдущего
  101   | 2024-08-30 10:13:00 | click
  102   | 2024-08-30 10:00:00 | page_view
  102   | 2024-08-30 10:01:00 | click

Результат (sessions):

user_id | session_id | session_start        | session_end          | event_count | duration_minutes
--------|------------|----------------------|----------------------|-------------|---------------
  101   |     1      | 2024-08-30 10:00:00 | 2024-08-30 10:04:30 |      3      |       4.5
  101   |     2      | 2024-08-30 10:12:00 | 2024-08-30 10:13:00 |      2      |       1
  102   |     1      | 2024-08-30 10:00:00 | 2024-08-30 10:01:00 |      2      |       1

Важные параметры

Таймаут 5 минут = 300 секунд

Другие популярные таймауты:

• 30 минут = 1800 секунд (веб-сессии)
• 15 минут = 900 секунд (мобильные приложения)
• 10 минут = 600 секунд (средний уровень)

Оптимизация для больших объёмов

-- Для петабайт данных рекомендуется добавить партиционирование
WITH events AS (
    SELECT * 
    FROM raw_events
    WHERE ds >= '2024-08-01' AND ds <= '2024-08-31'  -- Партиция по дате
    AND user_id IS NOT NULL -- Фильтруем невалидные
)

-- Остальной SQL как выше...

Проверка качества сессий

-- Проверить, есть ли пропуски в сессиях
WITH sessions AS (
    -- Предыдущий запрос...
)
SELECT 
    user_id,
    session_id,
    MIN(session_start) as first_event,
    MAX(session_end) as last_event,
    COUNT(*) as num_events,
    (unix_timestamp(MAX(session_end)) - unix_timestamp(MIN(session_start))) / 60 as total_duration_minutes
FROM sessions
GROUP BY user_id, session_id
HAVING COUNT(*) < 3  -- Показать сессии с менее чем 3 событиями
ORDER BY user_id;

Выводы

✅ Используйте оконные функции (LAG, SUM OVER) для определения границ сессий ✅ Таймаут 5 минут — стандарт для веб-аналитики ✅ Spark/SQL решение лучше масштабируется для Big Data ✅ Проверяйте качество результатов на выборке данных ✅ Партиционируйте по дате для оптимизации на больших объёмах