Привет! Меня зовут Сергей Прокопкин, я развиваю мотивационную программу и отвечаю за realtime-систему трекинга заданий в  Авито. 

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

На первый взгляд техническая задача выглядит довольно просто: получаем действия пользователя, проверяем их на соответствие активным заданиям и обновляем прогресс. Но когда поток событий становится большим, а число заданий и аудитория растут, простая схема начинает дорого обходиться. В нашем случае выяснилось, что более 99% обрабатываемых событий вообще не влияли на прогресс пользователей.

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

Содержание

• Как работал трекинг изначально

• Выносим фильтрацию перед основной системой

• FlinkSQL как бесконечный SQL-запрос

• Что нужно знать, чтобы отфильтровать событие

• Как мы добавляем к событиям актуальное состояние пользователя

• Динамически задерживаем поток

• Превращаем задания в SQL

• Куда отправлять результат

• Но задания постоянно меняются

• Что делать, если нужного поля нет в событии

• Сервис для обогащения событий 

• Фильтрация по обогащённым данным 

• Что получилось

• Что мы из этого вынесли

Как работал трекинг изначально

Путь пользовательского действия выглядел так: действие пользователя → Event Platform → Kafka → сервис мотивационной программы → фильтрация → обновление прогресса.

Основная логика фильтрации находилась непосредственно в сервисе мотивационной программы. После получения события из Kafka сервис:

1. Получал все активные задания пользователя;

2. Проверял, соответствует ли действие условиям какого-либо задания;

3. Если соответствовало — увеличивал прогресс и запускал дальнейшую логику награждения.

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

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

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

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

Второй вариант — использовать существующую потоковую платформу и описывать обработку декларативно. У этого подхода тоже есть цена: появляется зависимость от возможностей платформы и команды, которая её поддерживает, а разбор некоторых проблем становится сложнее. Зато значительная часть инфраструктурных задач уже решена. В итоге мы выбрали этот вариант. Так мы пришли к Flink и FlinkSQL.

Выносим фильтрацию перед основной системой

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

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

Второй вариант — использовать существующую потоковую платформу и описывать обработку декларативно. У этого подхода тоже есть цена: появляется зависимость от возможностей платформы и команды, которая её поддерживает, а разбор некоторых проблем становится сложнее. Зато значительная часть инфраструктурных задач уже решена. В итоге мы выбрали этот вариант. Так мы пришли к Flink и FlinkSQL.

FlinkSQL как бесконечный SQL-запрос

Чтобы дальше было проще понимать архитектуру, коротко разберу модель, с которой мы работаем.

В FlinkSQL Kafka-топик можно представить как таблицу. Сообщение в топике становится строкой, а поля JSON-сообщения — столбцами.

Но есть принципиальное отличие от обычной базы данных. SQL-запрос здесь не завершается после обработки существующих записей. Flink превращает его в непрерывно работающую job, которая читает новые события по мере их появления.

FlinkSQL Planner разбирает DDL- и DML-запросы, проверяет типы и строит план выполнения. Затем план превращается в JobGraph: чтение из Kafka, преобразования, фильтрация, запись результата. Flink разворачивает этот граф в кластере, после чего job продолжает работать с поступающим потоком.

Получается, что SQL описывает, что нужно сделать, а Flink превращает это описание в постоянно работающий обработчик событий.

Что нужно знать, чтобы отфильтровать событие

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

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

Для пайплайна фильтрации нам нужно было решить три задачи:

1. Дополнять события пользователя актуальным состоянием его заданий в мотивационной программе;

2. Автоматически обновлять пайплайн при изменении набора заданий;

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

Начнём с состояния. 

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

Чтобы проверить событие, SQL-фильтру нужны не только поля самого события, но и данные о заданиях пользователя. Поэтому перед фильтрацией отдельная Flink job дополняет события состоянием пользователя.

Запрашивать это состояние у сервиса мотивации для каждого события было бы дорого. Job хранит последнюю полученную версию во Flink State — механизме, который позволяет ей сохранять данные между событиями. 

Если состояния пользователя ещё нет, job запрашивает его у сервиса мотивации. Когда задания меняются, сервис сам инициирует обновление. При этом источником истины всегда остаётся сервис мотивации. 

События поступают во Flink job из входящего потока и обрабатываются независимо для каждого пользователя. Flink job хранит последний полученный от основной системы state каждого пользователя в своём кэше с TTL в несколько часов. 

При получении нового события job сначала проверяет его наличие в своем кэше. Если state найден, то в событие добавляется закэшированный state и отправляется дальше. Если state отсутствует (например, ещё не был загружен или истёк его TTL), job отправляет в основную систему запрос на обновление. До получения ответа события этого пользователя накапливаются в буфере. 

При этом ожидание state одного пользователя не останавливает обработку событий остальных пользователей. Пока для user_1 ожидается обновление state, события других пользователей, например user_2, продолжают обрабатываться. Если для user_2 state уже есть в кэше, его событие обрабатывается без ожидания user_1.

Все события, которые поступили для пользователя во время ожидания актуального state, накапливаются. Например, пока выполняется обновление state для user_1, событие event_1 ожидает актуальный state.

Когда основная система присылает актуальный state пользователя, job сохраняет его в своем кэше и добавляет его ко всем накопленным событиям пользователя. 

На первый взгляд задача решена. Но здесь возникает классическая проблема распределённых систем. Поток пользовательских событий и поток обновлений state идут независимо друг от друга. Событие и обновление состояния могут иметь разные задержки доставки.

В результате пользователь уже получил новое задание, совершил нужное действие, событие пришло во Flink — а новый state ещё находится в пути. Если сразу обработать событие, оно будет проверено по устаревшему состоянию. И наличие state в кэше само по себе ничего не гарантирует. Cache hit не означает, что данные актуальны.

Динамически задерживаем поток

Для определения актуальности мы используем несколько временных параметров:

• event timestamp — время совершения пользовательского действия;

• last state timestamp — временную метку последнего состояния, полученного от основной системы;

• state delivery delay — задержку доставки state до job.

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

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

Превращаем задания в SQL

После того как состояние пользователя стало доступно в потоке, для FlinkSQL оно фактически превратилось в ещё одно поле таблицы. Теперь можно использовать активные задания прямо в условии фильтрации.

Для каждого активного задания мы генерируем свой SELECT.

Здесь проверяются три вещи:

1. Задание активно у пользователя;

2. Событие соответствует целевому действию;

3. Параметры события соответствуют условиям задания.

После этого мы объединяем запросы через UNION ALL.

В итоге набор продуктовых правил превращается в один SQL-запрос.

Куда отправлять результат

Выходной Kafka-топик тоже описывается как таблица FlinkSQL.

После этого Flink проверяет каждое новое событие. Если оно соответствует хотя бы одному активному заданию, результат попадает в Kafka. Если не соответствует ни одному — до основной системы событие вообще не доходит.

Но задания постоянно меняются

Статический SQL здесь был бы бесполезен. В мотивационной программе постоянно запускаются новые задания, изменяются существующие и заканчиваются старые. Значит, фильтрация должна автоматически следовать за бизнес-конфигурацией.

Источником истины остаётся основная система. При изменении заданий мы:

1. Собираем новый SQL;

2. Создаём Flink job с актуальными DDL- и DML-запросами;

3. Переводим обработку на новую конфигурацию;

4. Останавливаем старую job.

Новую job запускаем до остановки старой. Если она не запустится из-за невалидной конфигурации, старая продолжит обрабатывать события и трекинг не остановится. Когда обе job некоторое время работают параллельно, событие, подходящее под обе версии правил, может дважды попасть в результирующий топик. На стороне Motivation есть дедупликация событий, поэтому прогресс не засчитывается повторно. Таким образом, изменение бизнес-правил автоматически приводит к обновлению пайплайна фильтрации без ручного редактирования SQL и релиза кода фильтрации.

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

Что делать, если нужного поля нет в событии

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

Основная система на этапе построения FlinkSQL запроса определяет, какие данные уже присутствуют в событии, а какие нужно запросить. После первичной фильтрации мы добавляем в событие служебное поле __enrich_request. В нем указываются объект и список полей, которые необходимо получить. 

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

Оба маршрута создаются через INSERT внутри одной Flink job.

Сервис для обогащения событий 

События из event_to_enrich читает отдельный сервис на Go. Он:

1. Получает событие и читает __enrich_request.

2. Определяет, какие данные нужны для проверки задания.

3. Собирает эти данные из других систем.

4. Добавляет результат обогащения к событию.

5. Отправляет событие в топик enriched events.

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

Но после обогащения событие всё ещё нужно проверить, и здесь мы снова используем FlinkSQL.

Фильтрация по обогащённым данным 

Первая job уже проверила все условия, которые можно было проверить по исходному событию, и записала идентификатор подходящего задания в __filtered_task_id. Поэтому второй фильтр проверяет только оставшиеся условия — те, которые зависят от __enrich_response. Если они выполняются, событие попадает в результирующий топик.

До основной системы в итоге доходят только события, прошедшие все необходимые фильтры.

Сначала событие поступает в User state Flink job. Она добавляет к нему состояние пользователя: активные задания и параметры, необходимые для проверки их условий. State кэшируется во Flink и обновляется сообщениями из основной системы. Если его нет, job запрашивает обновление и временно накапливает события этого пользователя. 

Затем событие поступает в FlinkSQL job 1. Здесь проверяются условия, доступные по исходным полям события и пользовательскому state: активно ли задание у пользователя, соответствует ли событие целевому действию и подходят ли его параметры. Если событие не соответствует ни одному заданию, оно отсеивается. Если все условия можно проверить на этом этапе, подходящее событие направляется в основную систему.

Если для окончательной проверки нужны дополнительные данные, к событию добавляется запрос на обогащение. Enrich Facade Service получает необходимые сведения из других систем и дополняет ими событие. После этого FlinkSQL job 2 проверяет оставшиеся условия задания по обогащённым данным. События, прошедшие эту проверку, также направляются в основную систему.

Основная система обрабатывает полученные события и обновляет прогресс заданий. Изменения состояния пользователя отправляются обратно в User state Flink job и учитываются при обработке следующих событий. А при изменении конфигурации самих заданий основная система генерирует новые SQL-запросы и обновляет фильтрующие jobs. 

Таким образом, итоговый поток выглядит следующим образом: User state → фильтрация по сырым данным → обогащение при необходимости → фильтрация по обогащённым данным → основная система → обновление user state.

Что получилось

После раскатки новой фильтрации в 600 раз уменьшился поток событий в основную систему, в 25 раз сократилось число запросов к шардированному Redis и на 30% снизилась нагрузка на основную PostgreSQL.

При этом более 99% событий, которые теперь доходят до основной системы, засчитываются в прогресс заданий.

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

Что мы из этого вынесли

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

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

Третий — динамический продукт требует динамического pipeline. Если задания постоянно создаются и изменяются, ручное обновление streaming-конфигурации быстро превращается в проблему. Поэтому изменение бизнес-правил должно автоматически приводить к обновлению обработки потока.