Aggregate таблица

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

В сценариях анализа и агрегации данных Aggregate таблицы могут уменьшить объем данных, которые необходимо обработать, тем самым повышая эффективность запросов.

Сценарии

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

• Помочь провайдерам веб-сайтов или приложений анализировать объем трафика и время, которое их пользователи проводят на конкретном веб-сайте или в приложении, и общее количество посещений веб-сайта или приложения.

• Помочь рекламным агентствам анализировать общее количество кликов, общее количество просмотров и статистику потребления рекламы, которую они предоставляют для своих клиентов.

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

Запрос данных и загрузка в вышеуказанных сценариях имеют следующие характеристики:

• Большинство запросов — это агрегированные запросы, такие как SUM, MAX и MIN.
• Необработанные детализированные данные не нужно извлекать.
• Исторические данные не обновляются часто. Добавляются только новые данные.

Принцип

От фазы загрузки данных до запроса данных данные в Aggregate таблицах агрегируются несколько раз следующим образом:

1. На этапе загрузки данных каждая партия данных формирует версию, когда данные загружаются в Aggregate таблицу партиями. В одной версии данные с одинаковым ключом агрегации будут агрегированы.

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

3. На этапе запроса данных Selena агрегирует данные, которые имеют одинаковый ключ агрегации среди всех версий данных, прежде чем вернуть результат запроса.

Операции агрегации помогают уменьшить объем данных, которые необходимо обработать, тем самым ускоряя запросы.

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

Date	Country	PV
2020.05.01	CHN	1
2020.05.01	CHN	2
2020.05.01	USA	3
2020.05.01	USA	4

Selena агрегирует четыре необработанные записи в следующие две записи при загрузке данных.

Date	Country	PV
2020.05.01	CHN	3
2020.05.01	USA	7

Создание таблицы

Предположим, что вы хотите проанализировать количество посещений пользователями из разных городов на различные веб-страницы. В этом примере создайте таблицу с именем example_db.aggregate_tbl, определите site_id, date и city_code в качестве ключа агрегации, определите pv в качестве столбца значений и укажите функцию SUM для столбца pv.

Инструкция для создания таблицы следующая:

CREATE TABLE aggregate_tbl (
    site_id LARGEINT NOT NULL COMMENT "id of site",
    date DATE NOT NULL COMMENT "time of event",
    city_code VARCHAR(20) COMMENT "city_code of user",
    pv BIGINT SUM DEFAULT "0" COMMENT "total page views"
)
AGGREGATE KEY(site_id, date, city_code)
DISTRIBUTED BY HASH(site_id);

NOTICE

• При создании таблицы вы должны указать столбец bucketing с использованием клаузы DISTRIBUTED BY HASH. Для получения подробной информации см. bucketing.
• Начиная с v1.5.2, Selena может автоматически устанавливать количество bucket (BUCKETS) при создании таблицы или добавлении partition. Вам больше не нужно вручную устанавливать количество bucket. Для получения подробной информации см. set the number of buckets.

Обобщенные состояния функций агрегации

Beta feature

Рекомендации по использованию бета-функций

Selena поддерживает обобщенные состояния функций агрегации начиная с v1.5.2.

В анализе и суммировании данных Aggregate таблицы уменьшают объем данных, обрабатываемых во время запросов, повышая производительность запросов. Для больших наборов данных Aggregate таблицы очень эффективны, поскольку они суммируют данные по измерениям перед запросом. Они также служат важным методом для инкрементного вычисления функций агрегации в Selena. Однако в более ранних версиях поддержка ограничена встроенными функциями, такими как SUM, MAX, MIN, REPLACE, HLL_UNION, PERCENTILE_UNION и BITMAP_UNION, в то время как теоретически все встроенные функции агрегации могут использоваться в aggregate таблицах. Для решения этого ограничения введены обобщенные состояния агрегации для поддержки хранения всех состояний встроенных функций.

Хранение обобщенных состояний агрегации

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

Определение:

col_name agg_func_name(parameter1_type, [parameter2_type], ...)

• col_name: Имя столбца.
• agg_func_name: Имя функции агрегации, промежуточные состояния которой необходимо сохранить.
• parameter_type: Тип входного параметра функции агрегации. Функция может быть однозначно идентифицирована с помощью типа параметра.

примечание

• Поддерживаются только встроенные функции Selena с по крайней мере одним параметром. Java и Python UDAF не поддерживаются.
• Для стабильности и расширяемости тип столбца состояния агрегации всегда Nullable (за исключением функции count) и не может быть изменен.
• Значения параметров не нужны для определения многопараметрических функций, поскольку типы могут быть выведены, а значения параметров не участвуют в вычислении.
• Сложные параметры, такие как ORDER BY и DISTINCT, не поддерживаются.
• Поддержка конкретных встроенных функций, таких как GROUP_CONCAT, WINDOW_FUNNEL и APPROX_TOP_K, все еще находится в разработке. Они будут поддерживаться в будущих релизах. Для получения подробной информации см. FunctionSet.java#UNSUPPORTED_AGG_STATE_FUNCTIONS.

Пример:

CREATE TABLE test_create_agg_table (
  dt VARCHAR(10),
  -- Define generic aggregate state storage.
  hll_sketch_agg ds_hll_count_distinct(varchar),
  avg_agg avg(bigint),
  array_agg_agg array_agg(int),
  min_by_agg min_by(varchar, bigint)
)
AGGREGATE KEY(dt)
PARTITION BY (dt)
DISTRIBUTED BY HASH(dt) BUCKETS 4;

Функции-комбинаторы

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

Функция-комбинатор _state

Функция _state преобразует входные параметры в типы промежуточного состояния.

Определение:

agg_intermediate_type {agg_func_name}_state(input_col1, [input_col2], ...)

• agg_func_name: Имя функции агрегации, которая должна передать входные параметры в типы промежуточного состояния.
• input_col1/col2: Входные столбцы функции агрегации.
• agg_intermediate_type: Тип возвращаемого значения функции _state, то есть тип промежуточного состояния функции агрегации.

примечание

_state — это скалярная функция. Вам не нужно определять столбец агрегации для вычисления состояния входного параметра.

Пример:

CREATE TABLE t1 (
  id BIGINT NOT NULL,
  province VARCHAR(64),
  age SMALLINT,
  dt VARCHAR(10) NOT NULL
)
DUPLICATE KEY(id)
PARTITION BY (dt)
DISTRIBUTED BY HASH(id) BUCKETS 4;

INSERT INTO t1 SELECT generate_series, generate_series, generate_series % 10, "2024-07-24" FROM table(generate_series(1, 100));

-- Transfer the data in t1 with _state combinator function, and insert it into the Aggregate table.
INSERT INTO test_create_agg_table
SELECT
    dt,
    ds_hll_count_distinct_state(id),
    avg_state(id),
    array_agg_state(id),
    min_by_state(province, id)
FROM t1;

Функция-комбинатор _union

Функция _union объединяет несколько столбцов промежуточного состояния в одно состояние.

Определение:

-- Union multiple aggregate intermediate states.
agg_intermediate_type {agg_func_name}_union(input_col)

• agg_func_name: Имя функции агрегации.
• input_col: Входные столбцы функции агрегации. Тип входного столбца — это тип промежуточного состояния функции агрегации. Вы можете получить его с помощью функций _state.
• agg_intermediate_type: Тип возвращаемого значения функции _union, то есть тип промежуточного состояния функции агрегации.

примечание

_union — это функция агрегации. Она возвращает тип промежуточного состояния вместо типа окончательного результата функции.

Пример:

-- Case 1: Union the intermediate states of the Aggregate table.
SELECT
    dt,
    ds_hll_count_distinct_union(hll_sketch_agg),
    avg_union(avg_agg),
    array_agg_union(array_agg_agg),
    min_by_union(min_by_agg)
FROM test_create_agg_table
GROUP BY dt
LIMIT 1;

-- Case 2: Union the intermediate states input by the _state combinator function.
SELECT
    dt,
    ds_hll_count_distinct_union(ds_hll_count_distinct_state(id)),
    avg_union(avg_state(id)),
    array_agg_union(array_agg_state(id)),
    min_by_union(min_by_state(province, id))
FROM t1
GROUP BY dt
LIMIT 1;

Функция-комбинатор _merge

Функция-комбинатор _merge инкапсулирует функции агрегации как обобщенную функцию агрегации для вычисления окончательного результата агрегации нескольких промежуточных состояний.

Определение:

-- Merge multiple aggregate intermediate states.
agg_result_type {agg_func_name}_merge(input_col)

• agg_func_name: Имя функции агрегации.
• input_col: Входные столбцы функции агрегации. Тип входного столбца — это тип промежуточного состояния функции агрегации. Вы можете получить его с помощью функций _state.
• agg_intermediate_type: Тип возвращаемого значения функции _merge, то есть окончательный результат агрегации функции агрегации.

Пример:

-- Case 1: Merge the intermediate states of the Aggregate table to obtain the final aggregation result.
SELECT
    dt,
    ds_hll_count_distinct_merge(hll_sketch_agg),
    avg_merge(avg_agg),
    array_agg_merge(array_agg_agg),
    min_by_merge(min_by_agg)
FROM test_create_agg_table
GROUP BY dt
LIMIT 1;

-- Case 2:  Merge the intermediate states input by the _state combinator function to obtain the final aggregation result.
SELECT
    dt,
    ds_hll_count_distinct_merge(ds_hll_count_distinct_state(id)),
    avg_merge(avg_state(id)),
    array_agg_merge(array_agg_state(id)),
    min_by_merge(min_by_state(province, id))
FROM t1
GROUP BY dt
LIMIT 1;

Использование обобщенных состояний агрегации в материализованных представлениях

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

Обобщенные состояния агрегации в синхронных материализованных представлениях

Пример:

-- Create a synchronous materialized view test_mv1 to store aggregate states.
CREATE MATERIALIZED VIEW test_mv1
AS
SELECT
    dt,
    -- Original aggregate functions.
    min(id) AS min_id,
    max(id) AS max_id,
    sum(id) AS sum_id,
    bitmap_union(to_bitmap(id)) AS bitmap_union_id,
    hll_union(hll_hash(id)) AS hll_union_id,
    percentile_union(percentile_hash(id)) AS percentile_union_id,
    -- Generic aggregate state functions.
    ds_hll_count_distinct_union(ds_hll_count_distinct_state(id)) AS hll_id,
    avg_union(avg_state(id)) AS avg_id,
    array_agg_union(array_agg_state(id)) AS array_agg_id,
    min_by_union(min_by_state(province, id)) AS min_by_province_id
FROM t1
GROUP BY dt;

-- Wait until rollup creation finishes.
show alter table rollup;

-- Direct queries against the aggregate function will be transparently accelerated by test_mv1.
SELECT
    dt,
    min(id),
    max(id),
    sum(id),
    bitmap_union_count(to_bitmap(id)), -- count(distinct id)
    hll_union_agg(hll_hash(id)), -- approx_count_distinct(id)
    percentile_approx(id, 0.5),
    ds_hll_count_distinct(id),
    avg(id),
    array_agg(id),
    min_by(province, id)
FROM t1
WHERE dt >= '2024-01-01'
GROUP BY dt;

-- Direct queries against the aggregate function and the rollup will also be transparently accelerated by test_mv1.
SELECT
    min(id),
    max(id),
    sum(id),
    bitmap_union_count(to_bitmap(id)), -- count(distinct id)
    hll_union_agg(hll_hash(id)), -- approx_count_distinct(id)
    percentile_approx(id, 0.5),
    ds_hll_count_distinct(id),
    avg(id),
    array_agg(id),
    min_by(province, id)
FROM t1
WHERE dt >= '2024-01-01';

DROP MATERIALIZED VIEW test_mv1;

Обобщенные состояния агрегации в асинхронных материализованных представлениях

Пример:

-- Create an asynchronous materialized view test_mv2 to store aggregate states.
CREATE MATERIALIZED VIEW test_mv2
PARTITION BY (dt)
DISTRIBUTED BY RANDOM
AS
SELECT
    dt,
    -- Original aggregate functions.
    min(id) AS min_id,
    max(id) AS max_id,
    sum(id) AS sum_id,
    bitmap_union(to_bitmap(id)) AS bitmap_union_id,
    hll_union(hll_hash(id)) AS hll_union_id,
    percentile_union(percentile_hash(id)) AS percentile_union_id,
    -- Generic aggregate state functions.
    ds_hll_count_distinct_union(ds_hll_count_distinct_state(id)) AS hll_id,
    avg_union(avg_state(id)) AS avg_id,
    array_agg_union(array_agg_state(id)) AS array_agg_id,
    min_by_union(min_by_state(province, id)) AS min_by_province_id
FROM t1
GROUP BY dt;

-- Refresh the materialized view.
REFRESH MATERIALIZED VIEW test_mv2 WITH SYNC MODE;

-- Direct queries against the aggregate function will be transparently accelerated by test_mv2.
SELECT
    dt,
    min(id),
    max(id),
    sum(id),
    bitmap_union_count(to_bitmap(id)), -- count(distinct id)
    hll_union_agg(hll_hash(id)), -- approx_count_distinct(id)
    percentile_approx(id, 0.5),
    ds_hll_count_distinct(id),
    avg(id),
    array_agg(id),
    min_by(province, id)
FROM t1
WHERE dt >= '2024-01-01'
GROUP BY dt;

SELECT
    min(id),
    max(id),
    sum(id),
    bitmap_union_count(to_bitmap(id)), -- count(distinct id)
    hll_union_agg(hll_hash(id)), -- approx_count_distinct(id)
    percentile_approx(id, 0.5),
    ds_hll_count_distinct(id),
    avg(id),
    array_agg(id),
    min_by(province, id)
FROM t1
WHERE dt >= '2024-01-01';

Примечания по использованию

• Ключ агрегации:

  • В инструкции CREATE TABLE ключ агрегации должен быть определен перед другими столбцами.

  • Ключ агрегации может быть явно определен с использованием AGGREGATE KEY. AGGREGATE KEY должен включать все столбцы, кроме столбцов значений, иначе таблица не будет создана.

    Если ключ агрегации не определен явно с использованием AGGREGATE KEY, все столбцы, кроме столбцов значений, считаются ключом агрегации по умолчанию.

  • Ключ агрегации имеет ограничение уникальности.

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

• Функция агрегации: Функция агрегации, используемая для столбца значений. Для поддерживаемых функций агрегации для Aggregate таблиц см. CREATE TABLE.

• Ключ сортировки

  • Начиная с v1.5.2, ключ сортировки отделен от ключа агрегации в Aggregate таблице. Aggregate таблица поддерживает указание ключа сортировки с использованием ORDER BY и указание ключа агрегации с использованием AGGREGATE KEY. Столбцы в ключе сортировки и ключе агрегации должны быть одинаковыми, но порядок столбцов не обязательно должен быть одинаковым.

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

• При создании таблицы вы можете создавать только индексы Bitmap или индексы Bloom Filter на ключевых столбцах таблицы.

Что делать дальше

После создания таблицы вы можете использовать различные методы загрузки данных для загрузки данных в Selena. Для получения информации о методах загрузки данных, поддерживаемых Selena, см. Loading options.

Note: Когда вы загружаете данные в таблицу, которая использует Aggregate таблицу, вы можете обновлять только все столбцы таблицы. Например, при обновлении вышеупомянутой таблицы example_db.aggregate_tbl вы должны обновить все ее столбцы, которые являются site_id, date, city_code и pv.