Реклама является одним из наиболее крупных сегментов практического применения технологий Big Data. Поэтому сегодня рассмотрим, как Flink SQL реализует потоковую аналитику больших данных в AdTech-кейсах. Разбираем пример JOIN-соединения двух потоков событий — показов и кликов, чтобы вычислить конверсию рекламной кампании средствами Apache Flink или Spark.
Потоки Big Data за фасадом контекстной рекламы: бизнес-контекст
В контекстной рекламе размещение объявления выполняется с помощью механизма, называемого назначением ставок в реальном времени (Real-Time Bidding). По сути, торги в реальном времени — это аукцион, на котором множество участников соревнуются за показ баннера или видео, называемых креативом (creative) конкретному конечному пользователю. Во время этого процесса DSP-платформы (Demand-Side Platforms) получают предложения по показу рекламы пользователям, идентифицированным по id их устройств, и отвечают ставками. Отслеживание показов рекламных объявлений и пользовательской реакции на них в виде кликов – ключевая задача контекстной рекламы.
Хотя процесс размещения контекстной рекламы в целом автоматизирован, иногда требуется ручной контроль. В частности, определение кампании и характеристик целевой аудитории (демографические данные, страна, критерии эффективности кампании). Также может потребоваться тщательный мониторинг эффективности кампании и корректировка некоторых параметров, особенно на ранней стадии после запуска, во время проверки гипотез.
Раньше задача анализа больших объемов данных решалась методами пакетной обработки. Однако, важно получать информацию в режиме реального времени, не ожидая часы, пока пакетное задание завершит обработку сырых данных. Потоковая аналитика помогает сохранить рекламный бюджет, корректируя параметры кампании. Кроме того, для любой метрики, основанной на корреляции следующих событий, пакетная обработка не даст корректных результатов, из-за ограничений размера пакетов и автономной обработки каждой порции данных разными заданиями.
Задача мониторинга кампании обычно выполняется маркетинговым или веб-аналитиком. Динамичный характер бизнеса можно требовать специальной интеграции с новыми потоками данных, добавления новых измерений к существующим потокам и других изменений. Поэтому для ускорения процессов получения результатов от контекстной рекламы желательно устранить зависимость аналитиков от дата-инженеров. Для этого нужен гибкий набор инструментов с низким порогом входа и простым внедрением, например, SQL. Этот структурированный язык запросов активно используется аналитиками и присутствует в Big Data фреймворках пакетных и потоковых вычислений, таких как Apache Spark и Flink. В частности, благодаря модулям SQL в Spark и Flink можно выполнять аналитики больших данных без разработки сложного кода на Java или Scala. Таким образом, можно анализировать большие объемы сырых данных «на лету» и легко создавать интерактивные пользовательских дэшборды в режиме самообслуживания.
Вычисление конверсии рекламной кампании средствами Apache Flink SQL
Итак, в рассматриваемом примере обрабатываются 2 потока данных: показов (представлений) рекламных объявлений и кликов на них. В Apache Flink они регистрируются как таблицы путем определения их схемы и параметров таблицы.
Поток показов (Impressions) состоит из событий, каждое из которых означает победу в аукционе с назначением ставок в реальном времени и успешную демонстрацию креатива пользователю. Событие включает такие данные, как размеры креатива (баннера или видео), код страны и идентификатор рекламной кампании.
Поток кликов (Clicks) связан с потоком показов. Поле Correlation_id в таблице кликов соответствует полю bid_id в таблице показов. Это будет основой для соединения потоков данных и расчета конверсии кликов (CTR). CTR равна отношению количества кликов к общему количеству выполненных показов.
Следующий SQL-запрос в Apache Flink создает одну строку для каждого показа и сопоставляет ее с кликом (при его наличии), который наблюдался в течение двух минут после показа объявления.
CREATE TEMPORARY VIEW impressions_with_clicks_raw AS
SELECT
i.bid_id,
i.campaign_id,
i.country_code,
i.creative_dimensions,
i.`event_time` AS serve_time,
c.tracker,
c.`timestamp` AS click_time,
CASE
WHEN c.`timestamp` IS NULL THEN FALSE
WHEN c.`timestamp` IS NOT NULL THEN TRUE
END AS clicked
FROM impressions i
LEFT OUTER JOIN clicks c
ON i.bid_id = c.correlation_id AND
c.event_time BETWEEN i.event_time AND
i.event_time + INTERVAL ‘2’ MINUTE ;
Здесь создается временное представление (impressions_with_clicks_raw) для объединения нескольких запросов. Это не приводит к фактическому выполнению задания, пока на него не ссылаются в другом месте. При использовании в составе представления другого запроса среда выполнения SQL-запросов Flink сгенерирует план его выполнения и выполнит оптимизацию, будто для одного вложенного запроса. Разумеется, временные представления не являются обязательными, те же результаты могут быть получены и без них, однако, этот прием повышает удобство чтения запроса.
Важно, что при использовании потокового SQL всегда следует ограничить время запроса, иначе внутреннее состояние базового приложения Flink будет бесконечно расти. Например, здесь предложение BETWEEN позволяет Flink удалять события из состояния, когда прошло 120 (интервал совпадения) + 5 (водяной знак таблицы) секунд времени события:
BETWEEN i.event_time AND i.event_time + INTERVAL ‘2’ MINUTE
Далее можно агрегировать необработанные данные и подсчитать количества показов с кликами и без них с разбивкой по нужным параметрам (идентификатор кампании, код страны и пр.). Уменьшить количество обновлений позволят оконные операции, определяющие период группировки и подсчета элементов, например, «кувыркающееся» окно (Tumble) длительностью 5 минут. Наконец, следует выполнить внутреннее соединение итоговой таблицы самой с собой (self join), чтобы рассчитать окончательный рейтинг кликов для каждой кампании для каждой страны:
CREATE TEMPORARY VIEW ctr_campaigns AS
SELECT
ic1.country_code,
ic1.campaign_id,
ic1.cnt AS `clicks_count`,
ic2.cnt AS `no_clicks_count`,
CAST(((100.0*ic1.cnt/(ic1.cnt + ic2.cnt)))
AS DECIMAL(8,4) ) AS ctr
FROM impressions_with_clicks_60s AS ic1
JOIN impressions_with_clicks_60s AS ic2
ON ic1.window_end = ic2.window_end AND
ic1.country_code = ic2.country_code AND
ic1.campaign_id = ic2.campaign_id AND
ic1.clicked = TRUE AND
ic2.clicked = FALSE;
Далее можно создать дэшборд для визуализации полученных результатов. Если требуется сделать обновления доступными для BI-систем, их следует сперва записать их в таблицу, поддерживаемую внешней базой данных. DDL-операторы Flink хранят только метаданные таблицы в каталоге, но не вызывают функций во внешних системах, таких как создание таблицы или индекса.
Например, следующий SQL-запрос показывает создание временной таблицы для записи данных из Apache Flink во внешнюю СУБД PostgreSQL:
CREATE TEMPORARY TABLE `ctr_dashboard` (
`country_code` VARCHAR(2),
`campaign_id` INT,
`clicks_count` BIGINT NOT NULL,
`no_clicks_count` BIGINT NOT NULL,
`ctr` DECIMAL(8, 4) NOT NULL,
PRIMARY KEY (`country_code`, `campaign_id`) NOT ENFORCED
)
WITH (
‘connector’ = ‘jdbc’,
‘url’ = ‘jdbc:postgresql://postgres.databases.svc:5432/adtech’,
‘table-name’ = ‘ctr_dashboard_campaigns’,
‘username’ = ‘flink’,
‘password’ = ‘12345’
);
Определение первичного ключа в полях country_code и campaign_id позволит отображать самые последние значения конверсии, обновляемые примерно каждые 5 минут. К этому моменту все входящие необработанные впечатления и события кликов будут оценены и преобразованы Flink в соответствующую метрику, чтобы визуализировать их в любом BI-инструменте.
Производительность и практическое применение
DSP-платформа в аукционе с назначением ставок в реальном времени, может получать трафик около 100 000 запросов в секунду. В зависимости от размера кампании и стратегии назначения ставок, это может привести к большому количеству показов в секунду. Apache Flink может легко обрабатывать этот большой объем «на лету», без непрестанных обращений к СУБД с сырыми данными. При этом аналитики могут применять привычный язык SQL-запросов.
Также стоит отметить четкое разделение Flink между обработкой данных, сохранением таблицы (через коннекторы) и отказоустойчивостью состояния (моментальные снимки в хранилище BLOB-объектов). Отказоустойчивость Flink SQL основана на механизме контрольных точек, который используется для прочих приложений Flink. Однако, в отличие от KSQL, Apache Flink не нуждается в дополнительных разделах в кластере для обеспечения отказоустойчивости выполняемых запросов. Таким образом, снижается нагрузка на системы, хранящие бизнес-данные. Этот подход отлично подходит для систем самообслуживания, где непросто создавать большие объемы активно реплицируемого состояния при отправке произвольного пользовательского запроса. Еще пару примеров подобного использования Apache Flink для пакетной и потоковой обработки данных читайте в нашей новой статье.
В заключение отметим, что подобную систему можно реализовать и средствами Apache Spark, который также имеет SQL-модуль и возможности потоковой обработки данных практически в реальном времени. Как это сделать на практике, а также освоить другие особенности разработки распределенных приложений аналитики больших данных с Apache Spark, Flink и другими компонентами экосистемы Hadoop, вы узнаете на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
