Непрерывные агрегатные функции

Одной из основных целей Tantor PipelineDB является обеспечение высокопроизводительной непрерывной агрегации данных, и поэтому агрегатные функции являются центральным компонентом инструмента Tantor PipelineDB. Непрерывные агрегатные функции могут быть очень мощным инструментом — они позволяют поддерживать постоянное количество данных, хранимых в Tantor PipelineDB, относительно количества данных, которые были переданы через него. Это может обеспечить устойчивую и очень высокую пропускную способность данных на аппаратном обеспечения с минимальными характеристиками.

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

Но для более сложных агрегатных функций, таких как avg, stddev, percentile_cont и т. д., требуется более продвинутая инфраструктура для поддержки эффективных инкрементальных обновлений, и Tantor PipelineDB прозрачно обрабатывает данные по сложным сценариям.

Ниже вы найдете описание всех агрегатных функций, которые поддерживаются в Tantor PipelineDB. Некоторые из них ведут себя немного иначе, чем их стандартные аналоги, что обусловлено необходимостью оптимизировать работу с бесконечными потоками данных. К таким агрегатным функциям идет пояснение о том, чем именно отличается их поведение.

Примечание

Для получения информации по агрегатным функциям смотрите документацию PostgreSQL.

Фильтры Блума

bloom_agg ( expression )

Добавляет все входные значения в фильтр Фильтр Блума.

bloom_agg ( expression , p , n )

Добавляет все входные значения в фильтр Блума и устанавливает его размер в соответствии с заданными параметрами. p — желаемый процент ложного срабатывания, а n — ожидаемое количество добавляемых уникальных элементов.

bloom_union_agg ( bloom filter )

Объединяет все входные фильтры Блума в один фильтр Блума, содержащий всю информацию из входных фильтров Блума.

bloom_intersection_agg ( bloom filter )

Берет пересечение всех входных фильтров Блума и объединяет их в один фильтр Блума, содержащий только общую информацию для всех входных фильтров Блума.

Смотрите раздел Функции фильтра Блума для получения информации о функционале, который можно использовать для работы с фильтрами Блума.

Отслеживание частоты

freq_agg ( expression )

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

freq_agg ( expression, epsilon, p )

То же, что и выше, но принимает epsilon и p в качестве параметров для базового cmsketch. epsilon определяет допустимую погрешность cmsketch и по умолчанию равен 0.002 (0.2%). p определяет достоверность и по умолчанию равен 0.995 (99.5%). Уменьшение значения epsilon и p приведет к более мелким структурам cmsketch, а увеличение наоборот — к более крупным.

freq_merge_agg ( count-min sketch )

Объединяет все входные структуры Count-min sketch в одну, содержащую всю информацию входных структур Count-min sketch.

Смотрите раздел Функции частоты для получения информации о функциях, которые можно использовать для работы со структурами Count-Min sketch.

Top-K

topk_agg ( expression , k )

Добавляет все входные значения в структуру данных Top-K, размер которой равен заданному k, увеличивая счетчик каждого значения на 1 каждый раз, когда добавляется значение.

topk_agg (expression, k, weight )

То же, что и выше, но связывает заданный вес с входным выражением (вместо веса по умолчанию, равного 1).

topk_merge_agg ( topk )

Объединяет все входные структуры Top-K в одну.

Смотрите раздел Функции Top-K для получения информации о функциях, которые можно использовать для работы с объектами Top-K.

HyperLogLog

hll_agg ( expression )

Добавляет все входные значения в HyperLogLog.

hll_agg ( expression, p )

Добавляет все входные значения в HyperLogLog с заданным p. Большее значение p уменьшает погрешность HyperLogLogs за счет увеличения размера.

hll_union_agg ( hyperloglog )

Берет все входные алгоритмы HyperLogLogs и создает один HyperLogLog, содержащий всю информацию входных HyperLogLogs.

Смотрите раздел Функции HyperLogLog для получения информации о функциях, которые можно использовать для работы с объектами HyperLogLog.

Распределение

dist_agg ( expression )

Добавляет все входные значения в T-Digest, чтобы отслеживать распределение всех входных выражений.

dist_agg ( expression, compression )

То же, что и выше, но создается базовый T-Digest с использованием заданного сжатия compression. Значение compression должно быть целым числом в диапазоне [20, 1000]. Большее значение для сжатия с большей точностью приведет к большему значению tdigest, чем меньшее значение tdigest с меньшим значением сжатия.

Смотрите раздел Функции распределения для получения информации о функциях, которые можно использовать для работы с объектами T-Digest.

Различные агрегатные функции

bucket_agg ( expression , bucket_id )

Добавляет 4-байтовые хеши каждого входного выражения в бакет с заданным bucket_id. Каждый хеш может быть указан только один раз в одном бакете в любой момент времени. Поэтому бакеты могут рассматриваться как уникальные наборы хешей входных выражений.

bucket_agg ( expression , bucket_id , timestamp )

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

Смотрите раздел Различные функции для получения информации о функциях, которые можно использовать для работы с объектами bucket_agg.

exact_count_distinct ( expression )

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

Важно

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

first_values ( n ) WITHIN GROUP (ORDER BY sort_expression)

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

Примечание

Смотрите также раздел Встроенные функции Tantor PipelineDB, где разъясняются некоторые функции Tantor PipelineDB для работы с фильтрами Блум, структурами Count-min sketch, HyperLogLogs и T-Digests, не относящиеся к агрегатным функциям. Также ознакомьтесь с разделом Вероятностные структуры данных и алгоритмы для получения дополнительной информации о том, что это такое и как его использовать.

keyed_max ( key, value )

Возвращает значение value, связанное с «самым высоким» ключом key.

keyed_min ( key, value )

Возвращает значение value, связанное с «самым низким» ключом key.

set_agg ( expression )

Добавляет все входные значения в набор.

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

Операция combine

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

Для этого типа операций, в Tantor PipelineDB есть специальный агрегатная функция combine:

combine ( aggregate column )

Берет агрегированный столбец и объединяет все значения в одно значение, как если бы все отдельные входные агрегатные функции были агрегированы один раз.

Примечание

combine работает только с агрегированными столбцами, которые принадлежат непрерывным представлениям.

Давайте рассмотрим на примере:

postgres=# CREATE VIEW v AS SELECT g::integer, AVG(x::integer) FROM stream GROUP BY g;
CREATE VIEW
postgres=# INSERT INTO stream (g, x) VALUES (0, 10), (0, 10), (0, 10), (0, 10), (0, 10);
INSERT 0 5
postgres=# INSERT INTO stream (g, x) VALUES (1, 20);
INSERT 0 1
postgres=# SELECT * FROM v;
 g |         avg
---+---------------------
 0 | 10.0000000000000000
 1 | 20.0000000000000000
(2 rows)

postgres=# SELECT avg(avg) FROM v;
         avg
---------------------
 15.0000000000000000
(1 row)

postgres=# -- Но не учитывался тот факт, что значение 10 весит намного больше,
postgres=# -- потому что оно было вставлено 5 раз, в то время как 20 было вставлено только один раз.
postgres=# -- combine() учитывает этот вес
postgres=#
postgres=# SELECT combine(avg) FROM v;
       combine
---------------------
 11.6666666666666667
(1 row)

postgres=# -- Готово! Это та же средняя величина, которая была бы получена, если бы было запущено
postgres=# -- одно среднее значение всех 6 вышеуказанных вставленных значений. Но понадобилось
postgres=# -- всего две строки, чтобы это сделать.

Общие агрегатные функции

array_agg ( expression )

Входные значения, включая нули, объединенные в массив.

avg ( expression )

Среднее всех входных значений.

bit_and ( expression )

Побитовое AND всех ненулевых входных значений или значение null, если их нет.

bit_or ( expression )

Побитовое OR всех ненулевых входных значений или значение null, если их нет.

bool_and ( expression )

Истина, если все входные значения true, в противном случае — false.

bool_or ( expression )

Истина, если хотя бы одно входное значение true, в противном случае — false.

count ( * )

Количество строк ввода.

count ( DISTINCT expression)

Количество строк, для которых expression является уникальным.

Примечание

Подсчет точного числа выражений в бесконечном потоке потребовал бы бесконечной памяти, поэтому непрерывные представления используют HyperLogLog для выполнения подсчета точных значений с использованием постоянного объема данных и за постоянное время за счет допущения небольшой погрешности. Исходя из опыта, реализация HyperLogLog в Tantor PipelineDB имеет погрешность около 0,81%. Например, count distinct может выдать 1008, когда фактическое количество уникальных выражений было 1000.

count ( expression )

Количество строк, для которых expression не нулевое.

every ( expression )

Эквивалент bool_and.

json_agg ( expression )

Агрегирует значения как массив JSON.

json_object_agg ( key, value )

Агрегирует пары ключ-значение в виде JSON-объекта.

jsonb_agg ( expression )

Агрегирует значения как массив JSONB.

jsonb_object_agg ( key, value )

Агрегирует пары ключ-значение в виде объекта JSONB.

max ( expression )

Максимальное значение выражения среди всех входных значений.

min ( expression )

Минимальное значение выражения среди всех входных значений.

string_agg ( expression, delimiter )

Входные значения объединены в строку, разделенную разделителем delimiter.

sum ( expression )

Сумма expression по всем входным значениям.

Статистические агрегатные функции

corr ( y, x )

Коэффициент корреляции.

covar_pop ( y, x )

Ковариация совокупности.

covar_samp ( y, x )

Ковариация выборки.

regr_avgx ( y, x )

Среднее значение независимой переменной (sum(x)/N).

regr_avgy ( y, x )

Среднее значение независимой переменной (sum(y)/N).

regr_count ( y, x )

Количество строк ввода, в которых оба выражения не нулевые.

regr_intercept ( y, x )

Точка пересечения c осью Y линии, полученной методом наименьших квадратов, по данным (x, y).

regr_r2 ( y, x )

Квадрат коэффициента корреляции.

regr_slope ( y, x )

Наклон линии, полученной методом наименьших квадратов по данным (x, y).

regr_sxx ( y, x )

sum(X^2) - sum(X)^2/N — сумма квадратов независимой переменной.

regr_sxy ( y, x )

sum(X*Y) - sum(X) * sum(Y)/N — сумма произведений независимых и зависимых переменных.

regr_syy ( y, x )

sum(Y^2) - sum(Y)^2/N — сумма квадратов независимой переменной.

stddev ( expression )

Стандартное отклонение по выборке входных значений.

stddev_pop ( expression )

Стандартное отклонение по совокупности входных значений.

variance ( expression )

Дисперсия для выборки входных значений (квадрат стандартного отклонения по выборке).

var_pop ( expression )

Дисперсия для совокупности входных значений (квадрат стандартного отклонения по совокупности).

Агрегатные функции с упорядоченным набором

Агрегатные функции с упорядоченным набором применяют сортировку к вводу для получения соответствующих результатов, поэтому они используют предложение WITHIN GROUP. Синтаксис следующий:

aggregate_name ( [ expression [ , ... ] ] ) WITHIN GROUP ( order_by_clause )

Давайте посмотрим на несколько примеров.

Вычислите 99-й процентиль value:

SELECT percentile_cont(0.99) WITHIN GROUP (ORDER BY value) FROM some_table;

Или с непрерывным представлением:

CREATE VIEW percentile AS
SELECT percentile_cont(0.99) WITHIN GROUP (ORDER BY value::float8)
FROM some_stream;

percentile_cont ( fraction )

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

percentile_cont ( array of fractions )

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

Примечание

Вычисление процентилей на бесконечных потоках потребовало бы бесконечной памяти, поэтому обе формы percentile_cont, используемые непрерывными представлениями, используют T-Digest как способ оценки процентилей с очень высокой степенью точности. В общем, процентили в непрерывных представлениях более точны, если они ближе к верхним или нижним границам [0, 1).

Агрегатные функции с гипотетическим набором

Агрегатные функции с гипотетическим набором берут выражение и вычисляют какое-то значение в контексте набора входных строк. Например, rank(2) вычисляет ранг (rank) из значения 2 с учетом того, какими будут входные строки в конечном итоге.

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

aggregate_name ( [ expression [ , ... ] ] ) WITHIN GROUP ( order_by_clause )

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

CREATE VIEW continuous_rank AS
SELECT rank(42) WITHIN GROUP (ORDER BY value::float8)
FROM some_stream;

Это непрерывное представление будет непрерывно обновлять ранг 42, учитывая все события, которые оно прочитало.

rank ( arguments )

Ранг гипотетической строки с пропусками вместо дублирующихся строк.

dense_rank ( arguments )

Ранг гипотетической строки без пропусков.

Примечание

Вычисление гипотетического значения dense_rank при наличии бесконечного потока значений потребовало бы бесконечной памяти, поэтому непрерывные представления используют HyperLogLog для выполнения этой операции за постоянное время и используют постоянный объем данных, за счет небольшой погрешности. Исходя из опыта, реализация HyperLogLog в Tantor PipelineDB проходит с погрешностью около 0.2%. Другими словами, dense_rank (1000) в непрерывном представлении может показать 998, когда фактическое количество уникальных значений с более низким рангом было 1000.

percent_rank ( arguments )

Относительный ранг гипотетической строки, варьирующийся от 0 до 1.

cume_dist ( arguments )

Относительный ранг гипотетической строки, варьирующийся от 1/N до 1.

Неподдерживаемые агрегатные функции

mode ( )

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

percentile_disc ( arguments )

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

xmlagg ( xml )

К сожалению, не поддерживается.

<aggregate_name> (DISTINCT expression)

Только агрегатная функция count поддерживается с выражением DISTINCT, как указано выше в разделе Общие агрегатные функции. В будущих версиях возможно будет доступно использование Фильтр Блума, чтобы разрешить выражения DISTINCT для всех агрегатных функций.