Таймеры в YTsaurus Flow

Зачем нужны таймеры

Многие задачи потоковой обработки требуют ожидания: «если в течение 30 минут после показа не пришёл клик — считать конверсию несостоявшейся». Стандартный обмен сообщениями между компьютейшенами для этого не подходит: сообщение либо есть, либо нет, а ждать «отсутствия» события нельзя.

Таймеры решают эту задачу. Process function в transform-режиме может создать таймер — сказать системе «разбуди меня, когда время X наступит». Когда момент наступает, функция получает вызов ProcessTimer / on_timer / onTimer и может принять решение на основе накопленного стейта.

Типичные сценарии применения:

• Джойн с ожиданием: коррелировать показ рекламы с кликом, который может прийти с задержкой.
• Таймаут: обработать ситуацию, когда ожидаемое событие так и не пришло.
• Агрегации с окнами: закрыть временное окно и выдать результат, когда накопленные данные «протухают».

Как работают таймеры

Жизненный цикл

1. Регистрация. Компьютейшен вызывает output.add_timer / output.addTimer / output->AddTimer, передавая TriggerTimestamp и EventTimestamp.
2. Хранение. Таймер надёжно сохраняется в YTsaurus и дополнительно кешируется в памяти процесса.
3. Срабатывание. Когда EventWatermark (или иной сконфигурированный вотермарк) превышает TriggerTimestamp, таймер передаётся в компьютейшен.
4. Обработка и удаление. Результат обработки и удаление таймера фиксируются в одной транзакции — это обеспечивает гарантии exactly-once (см. Дедупликация).

Поля таймера

Каждый таймер содержит два временных поля:

• TriggerTimestamp — момент, когда таймер должен сработать. Временна́я шкала (event_time, system_time, real_time) задаётся в конфигурации.
• EventTimestamp — бизнес-время исходного события, «придержанное» в таймере. Позволяет при обработке таймера знать, когда произошло инициирующее событие.

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

Какие компьютейшены поддерживают таймеры

Конфигурация

Чтобы компьютейшен мог использовать таймеры, в его спеке необходимо заполнить поле timers. Каждый элемент массива описывает один таймер-стрим со следующими параметрами:

Источник: yt/yt/flow/library/cpp/common/spec.h

Пояснения:

• time_type определяет шкалу, по которой система сравнивает TriggerTimestamp с текущим вотермарком:

  • event_time (по умолчанию) — сравнивается с EventWatermark по всем input-потокам (или по потокам, указанным в streams).
  • system_time — сравнивается с SystemWatermark.
  • real_time — сравнивается с реальным астрономическим временем.

• streams / streams_with_delays — позволяют ограничить, какие входные потоки участвуют в расчёте вотермарка для данного таймера. streams_with_delays дополнительно позволяет задать индивидуальную задержку для каждого потока.

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

Порядок обработки

Внутри одного таймер-потока таймеры одного ключа обрабатываются в порядке TriggerTimestamp.

Между разными таймер-потоками порядок обработки не гарантирован. Чтобы таймер A сработал после таймера B, добавьте поток B в поле streams спецификации таймера A. Тогда A будет ожидать продвижения вотермарка потока B.

Блокировка связанных таймер-потоков

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

Таймер-потоки входят в одну группу, если у них одинаковый time_type (event_time или system_time) и пересекаются наборы потоков, по которым они ждут продвижения вотермарка. Группировка транзитивна: потоки, ожидающие {a, b}, {b, c} и {c, d}, образуют одну группу. Различия значений streams_with_delays не разделяют группу. Все таймер-потоки с real_time образуют отдельную группу независимо от настроенных входных потоков и задержек.

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

Структура таймера

Источник: yt/yt/flow/library/cpp/common/timer-inl.h

API по языкам

C++

void ProcessMessage(
    const TInputMessageConstPtr& message,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& /*context*/) override
{
    output->AddTimer(
        TSystemTimestamp(message->EventTimestamp.Underlying() + TDuration::Minutes(30).Seconds()),
        message->EventTimestamp);
}

void ProcessTimer(
    const TInputTimerConstPtr& timer,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context) override
{
    // timer->Key, timer->EventTimestamp, timer->TriggerTimestamp
    auto builder = context->MakeOutputMessageBuilder();
    // ...
    output->AddMessage(builder.Finish());
}

Подробнее — в разделе Process function (C++).

Java

// Создать таймер в onMessage:
output.addTimer(message.getEventTimestamp() + 30 * 60_000_000_000L, message.getEventTimestamp());

// Обработать сработавший таймер (RowFunction):
@Override
public void onTimer(Timer timer, OutputCollector output, RuntimeContext ctx) {
    // timer.getKey(), timer.getEventTimestamp(), timer.getTriggerTimestamp()
}

// Для BatchFunction:
@Override
public void onTimers(List<Timer> timers, OutputCollector output, RuntimeContext ctx) { ... }

Подробнее — в разделе Computation (Java).

Python

# Создать таймер в on_message:
output.add_timer(trigger_timestamp=message.event_timestamp + 30 * 60_000_000_000, event_timestamp=message.event_timestamp)

# Обработать сработавший таймер (RowFunction):
def on_timer(self, timer, output, ctx):
    # timer.key, timer.event_timestamp, timer.trigger_timestamp, timer.stream_id

# Для BatchFunction:
def on_timers(self, timers, output, ctx):
    for timer in timers:
        ...

Подробнее — в разделе Computation (Python).

Go

// Создать таймер в OnMessage:
out.AddTimer(flow.TimerRequest{
    TriggerTimestamp: msg.EventTimestamp + 30*60_000_000_000,
    EventTimestamp:   msg.EventTimestamp,
})

// Обработать сработавший таймер (RowFunction):
func (f myFunction) OnTimer(
    ctx context.Context,
    rt flow.Runtime,
    timer flow.Timer,
    out flow.OutputCollector,
) error {
    // timer.Key, timer.EventTimestamp, timer.TriggerTimestamp, timer.StreamID
    return nil
}

// Для BatchFunction:
func (f myFunction) OnTimers(ctx context.Context, rt flow.Runtime, timers []flow.Timer, out flow.OutputCollector) error { ... }

Подробнее — в разделе Computation (Go).

Примеры

Пример реализации джойна с ожиданием (показ + клик, таймаут 30 минут):

• C++
• Java
• Python
• Go

Ограничения и известные проблемы

• Swift-map режим не поддерживает таймеры. Если нужна работа с таймерами, запускайте process function под TProcessFunctionComputation.
• Временны́е метки передаются в наносекундах (uint64).

См. также

• Watermarks и Timers
• Stateful processing
• Computation (C++)
• Computation (Java)
• Computation (Python)
• Computation (Go)