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

Для представления этих данных будут использоваться три таблицы Postgres Pro. Для начала загрузите пример с данными для этих таблиц:

curl https://examples.citusdata.com/tutorial/companies.csv > companies.csv
curl https://examples.citusdata.com/tutorial/campaigns.csv > campaigns.csv
curl https://examples.citusdata.com/tutorial/ads.csv > ads.csv

Теперь можно дать указание citus распределить созданные таблицы по различным узлам кластера. Для этого запустите функцию create_distributed_table и укажите таблицу для сегментирования и столбец, по которому оно будет выполняться. В приведённом ниже примере все таблицы сегментируются по столбцу company_id.

SELECT create_distributed_table('companies', 'id');
SELECT create_distributed_table('campaigns', 'company_id');
SELECT create_distributed_table('ads', 'company_id');

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

Затем можно продолжить загрузку данных в таблицы с помощью стандартной команды psql \copy. Убедитесь, что указан правильный путь к файлу, если он был загружен не в стандартный каталог загрузки.

\copy companies from 'companies.csv' with csv
\copy campaigns from 'campaigns.csv' with csv
\copy ads from 'ads.csv' with csv

После завершения загрузки данных в таблицы можно выполнить несколько запросов. Расширение citus поддерживает стандартные команды INSERT, UPDATE и DELETE для вставки и изменения строк в распределённой таблице, что является одним из самых частых примеров взаимодействия пользователей с приложениями.

Например, можно добавить новую компанию, выполнив:

INSERT INTO companies VALUES (5000, 'New Company', 'https://randomurl/image.png', now(), now());

Чтобы удвоить бюджет всех кампаний предприятия, выполните команду UPDATE:

UPDATE campaigns
SET monthly_budget = monthly_budget*2
WHERE company_id = 5;

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

BEGIN;
DELETE FROM campaigns WHERE id = 46 AND company_id = 5;
DELETE FROM ads WHERE campaign_id = 46 AND company_id = 5;
COMMIT;

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

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

В этом разделе показано, как создать базу данных для приложения, анализирующего данные в реальном времени. Это приложение будет вставлять большие объёмы данных о событиях и выполнять аналитические запросы к этим данным с задержкой менее секунды. В этом примере используется набор данных событий Github, включающий в себя все общедоступные события на GitHub, такие как commit, fork, new issue и comment.

Для представления этих данных используются две таблицы Postgres Pro. Для начала загрузите образцы данных для этих таблиц:

curl https://examples.citusdata.com/tutorial/users.csv > users.csv
curl https://examples.citusdata.com/tutorial/events.csv > events.csv

Для начала подключитесь к узлу-координатору citus с помощью psql.

Если citus установлен согласно описанию в разделе Установка citus на одном узле, узел-координатор запускается и слушает порт 9700.

psql -p 9700

Затем можно создать таблицы стандартной командой Postgres Pro CREATE TABLE:

CREATE TABLE github_events
(
    event_id bigint,
    event_type text,
    event_public boolean,
    repo_id bigint,
    payload jsonb,
    repo jsonb,
    user_id bigint,
    org jsonb,
    created_at timestamp
);

CREATE TABLE github_users
(
    user_id bigint,
    url text,
    login text,
    avatar_url text,
    gravatar_id text,
    display_login text
);

Далее можно создать индексы данных о событиях так же, как это делается в Postgres Pro. В этом примере также показано, как создать индекс GIN, чтобы ускорить обращение к полям JSONB.

CREATE INDEX event_type_index ON github_events (event_type);
CREATE INDEX payload_index ON github_events USING GIN (payload jsonb_path_ops);

Теперь можно дать указание citus распределить созданные таблицы по узлам кластера. Для этого используйте функцию create_distributed_table и укажите таблицу для сегментирования и столбец, по которому оно будет выполняться. В приведённом ниже примере все таблицы сегментированы по столбцу user_id.

SELECT create_distributed_table('github_users', 'user_id');
SELECT create_distributed_table('github_events', 'user_id');

Сегментирование всех таблиц по столбцу user_id позволяет citus совмещать таблицы и эффективно использовать соединения и распределённые наборы группирования.

Затем можно продолжить загрузку данных в таблицы с помощью стандартной команды psql \copy. Убедитесь, что указан правильный путь к файлу, если он был загружен не в стандартный каталог загрузки.

\copy github_users from 'users.csv' with csv
\copy github_events from 'events.csv' with csv

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

SELECT count(*) FROM github_users;

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

SELECT date_trunc('minute', created_at) AS minute,
       sum((payload->>'distinct_size')::int) AS num_commits
FROM github_events
WHERE event_type = 'PushEvent'
GROUP BY minute
ORDER BY minute;

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

SELECT login, count(*)
FROM github_events ge
JOIN github_users gu
ON ge.user_id = gu.user_id
WHERE event_type = 'CreateEvent' AND payload @> '{"ref_type": "repository"}'
GROUP BY login
ORDER BY count(*) DESC LIMIT 10;

Расширение citus также поддерживает стандартные команды INSERT, UPDATE и DELETE для вставки и изменения данных. Например, можно изменить отображаемое имя пользователя, выполнив следующую команду:

UPDATE github_users SET display_login = 'no1youknow' WHERE user_id = 24305673;

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

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

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

В примере используются три сервиса:

Для начала подключитесь к узлу-координатору citus с помощью psql.

Если citus установлен согласно описанию в разделе Установка citus на одном узле, узел-координатор запускается и слушает порт 9700.

psql -p 9700

Теперь можно создавать роли базы данных для каждого микросервиса:

CREATE USER user_service;
CREATE USER time_service;
CREATE USER ping_service;

В citus схему можно распределять двумя способами:

Можно вывести список схем, распределённых на данный момент:

SELECT * FROM citus_shards;

schema_name  | colocation_id | schema_size | schema_owner
-------------+---------------+-------------+--------------
user_service |             5 | 0 bytes     | user_service
time_service |             6 | 0 bytes     | time_service
ping_service |             7 | 0 bytes     | ping_service
(3 rows)

Теперь нужно подключить каждый микросервис к узлу-координатору citus. Можно использовать команду \c для замены пользователя в текущем сеансе psql.

\c citus user_service

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    email VARCHAR(255) NOT NULL
);

\c citus time_service

CREATE TABLE query_details (
    id SERIAL PRIMARY KEY,
    ip_address INET NOT NULL,
    query_time TIMESTAMP NOT NULL
);

\c citus ping_service

CREATE TABLE ping_results (
    id SERIAL PRIMARY KEY,
    host VARCHAR(255) NOT NULL,
    result TEXT NOT NULL
);

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

git clone https://github.com/citusdata/citus-example-microservices.git

Репозиторий содержит следующие микросервисы: ping, time и user. Для каждого из них есть файл запуска app.py.

$ tree
.
├── LICENSE
├── README.md
├── ping
│   ├── app.py
│   ├── ping.sql
│   └── requirements.txt
├── time
│   ├── app.py
│   ├── requirements.txt
│   └── time.sql
└── user
    ├── app.py
    ├── requirements.txt
    └── user.sql

Перед запуском микросервисов отредактируйте файлы user/app.py, ping/app.py и time/app.py, предоставляющие конфигурации подключений для кластера citus:

# Database configuration
db_config = {
    'host': 'localhost',
    'database': 'citus',
    'user': 'ping_service',
    'port': 9700
}

После внесения изменений сохраните все отредактированные файлы и переходите к следующему этапу запуска микросервисов.

После вызова указанных выше функций API данные были сохранены, и можно проверить, соответствует ли представление citus_schemas ожидаемым результатам:

SELECT * FROM citus_shards;

schema_name  | colocation_id | schema_size | schema_owner
--------------+---------------+-------------+--------------
user_service |             1 | 112 kB      | user_service
time_service |             2 | 32 kB       | time_service
ping_service |             3 | 32 kB       | ping_service
(3 rows)

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

SELECT nodename,nodeport, table_name, pg_size_pretty(sum(shard_size))
  FROM citus_shards
GROUP BY nodename,nodeport, table_name;

nodename  | nodeport |         table_name         | pg_size_pretty
-----------+----------+----------------------------+----------------
localhost |     9701 | time_service.query_details | 32 kB
localhost |     9702 | user_service.users         | 112 kB
localhost |     9702 | ping_service.ping_results  | 32 kB

Видно, что микросервис time оказался на узле localhost:9701, а user и ping — на втором рабочем узле localhost:9702. Это всего лишь пример, и размерами данных здесь можно пренебречь, но лучше равномерно использовать пространство хранения между узлами. Разумнее разместить два меньших микросервиса time и ping на одном компьютере, а большой микросервис user — отдельно.

Это можно сделать, дав указание citus выполнить перебалансировку кластера по размерам дисков:

SELECT citus_rebalance_start();

NOTICE:  Scheduled 1 moves as job 1
DETAIL:  Rebalance scheduled as background job
HINT:  To monitor progress, run: SELECT * FROM citus_rebalance_status();
 citus_rebalance_start
-----------------------
                     1
(1 row)

После выполнения проверьте структуру:

SELECT nodename,nodeport, table_name, pg_size_pretty(sum(shard_size))
  FROM citus_shards
GROUP BY nodename,nodeport, table_name;

nodename  | nodeport |         table_name         | pg_size_pretty
-----------+----------+----------------------------+----------------
localhost |     9701 | time_service.query_details | 32 kB
localhost |     9701 | ping_service.ping_results  | 32 kB
localhost |     9702 | user_service.users         | 112 kB
(3 rows)

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

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

Рассмотрим упрощённую схему этого приложения. Оно должно отслеживать несколько предприятий, каждое из которых проводит рекламные кампании. В кампаниях много объявлений, и с каждым объявлением связаны записи о переходах и показах.

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

CREATE TABLE companies (
  id bigserial PRIMARY KEY,
  name text NOT NULL,
  image_url text,
  created_at timestamp without time zone NOT NULL,
  updated_at timestamp without time zone NOT NULL
);

CREATE TABLE campaigns (
  id bigserial PRIMARY KEY,
  company_id bigint REFERENCES companies (id),
  name text NOT NULL,
  cost_model text NOT NULL,
  state text NOT NULL,
  monthly_budget bigint,
  blacklisted_site_urls text[],
  created_at timestamp without time zone NOT NULL,
  updated_at timestamp without time zone NOT NULL
);

CREATE TABLE ads (
  id bigserial PRIMARY KEY,
  campaign_id bigint REFERENCES campaigns (id),
  name text NOT NULL,
  image_url text,
  target_url text,
  impressions_count bigint DEFAULT 0,
  clicks_count bigint DEFAULT 0,
  created_at timestamp without time zone NOT NULL,
  updated_at timestamp without time zone NOT NULL
);

CREATE TABLE clicks (
  id bigserial PRIMARY KEY,
  ad_id bigint REFERENCES ads (id),
  clicked_at timestamp without time zone NOT NULL,
  site_url text NOT NULL,
  cost_per_click_usd numeric(20,10),
  user_ip inet NOT NULL,
  user_data jsonb NOT NULL
);

CREATE TABLE impressions (
  id bigserial PRIMARY KEY,
  ad_id bigint REFERENCES ads (id),
  seen_at timestamp without time zone NOT NULL,
  site_url text NOT NULL,
  cost_per_impression_usd numeric(20,10),
  user_ip inet NOT NULL,
  user_data jsonb NOT NULL
);

Можно внести в схему изменения, которые повысят её производительность в распределённой среде, такой как citus. Чтобы понять, как это сделать, нужно ознакомиться с тем, как расширение распределяет данные и выполняет запросы.

Реляционная модель данных отлично подходит для приложений. Она защищает целостность данных, обеспечивает гибкость запросов и учитывает изменение данных. Традиционно считалось, что реляционные базы данных не могут масштабироваться до рабочих объёмов больших SaaS-приложений. Чтобы достичь такого размера, разработчикам приходилось прибегать к базам данных NoSQL или наборам серверных служб.

С помощью citus можно сохранить свою модель данных и сделать её масштабируемой. Расширение отображается в приложениях как единая база данных Postgres Pro, но внутри него запросы направляются на настраиваемое количество физических серверов (узлов), которые могут обрабатывать запросы параллельно.

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

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

В citus это можно сделать, убедившись, что в каждой таблице схемы есть столбец, позволяющий чётко обозначить, какому арендатору какие строки принадлежат. В приложении для аналитики рекламы арендаторами являются предприятия, поэтому необходимо убедиться, что во всех таблицах есть столбец company_id.

Можно дать указание citus использовать этот столбец для чтения и записи строк в один и тот же узел, если строки помечены для одного и того же предприятия. В терминологии citus company_id — это столбец распределения, который подробно описан в разделе Выбор столбца распределения.

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

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

В SQL такое требование означает объединение первичных и внешних ключей путём включения company_id. Это применимо и для сценария с несколькими арендаторами, поскольку в этом случае необходимо обеспечить уникальность для каждого арендатора.

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

CREATE TABLE companies (
  id bigserial PRIMARY KEY,
  name text NOT NULL,
  image_url text,
  created_at timestamp without time zone NOT NULL,
  updated_at timestamp without time zone NOT NULL
);

CREATE TABLE campaigns (
  id bigserial,       -- было: PRIMARY KEY
  company_id bigint REFERENCES companies (id),
  name text NOT NULL,
  cost_model text NOT NULL,
  state text NOT NULL,
  monthly_budget bigint,
  blacklisted_site_urls text[],
  created_at timestamp without time zone NOT NULL,
  updated_at timestamp without time zone NOT NULL,
  PRIMARY KEY (company_id, id) -- добавлено
);

CREATE TABLE ads (
  id bigserial,       -- было: PRIMARY KEY
  company_id bigint,  -- добавлено
  campaign_id bigint, -- было: REFERENCES campaigns (id)
  name text NOT NULL,
  image_url text,
  target_url text,
  impressions_count bigint DEFAULT 0,
  clicks_count bigint DEFAULT 0,
  created_at timestamp without time zone NOT NULL,
  updated_at timestamp without time zone NOT NULL,
  PRIMARY KEY (company_id, id),         -- добавлено
  FOREIGN KEY (company_id, campaign_id) -- добавлено
    REFERENCES campaigns (company_id, id)
);

CREATE TABLE clicks (
  id bigserial,        -- было: PRIMARY KEY
  company_id bigint,   -- добавлено
  ad_id bigint,        -- было: REFERENCES ads (id),
  clicked_at timestamp without time zone NOT NULL,
  site_url text NOT NULL,
  cost_per_click_usd numeric(20,10),
  user_ip inet NOT NULL,
  user_data jsonb NOT NULL,
  PRIMARY KEY (company_id, id),      -- добавлено
  FOREIGN KEY (company_id, ad_id)    -- добавлено
    REFERENCES ads (company_id, id)
);

CREATE TABLE impressions (
  id bigserial,         -- было: PRIMARY KEY
  company_id bigint,    -- добавлено
  ad_id bigint,         -- было: REFERENCES ads (id),
  seen_at timestamp without time zone NOT NULL,
  site_url text NOT NULL,
  cost_per_impression_usd numeric(20,10),
  user_ip inet NOT NULL,
  user_data jsonb NOT NULL,
  PRIMARY KEY (company_id, id),       -- добавлено
  FOREIGN KEY (company_id, ad_id)     -- добавлено
    REFERENCES ads (company_id, id)
);

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

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

Любые запросы приложений или операторы изменений, которые включают фильтр по company_id, будут продолжать работать в прежнем виде. Как упоминалось ранее, фильтры такого типа часто встречаются в многоарендных приложениях. При использовании объектно-реляционного преобразователя (ORM) можно распознавать эти запросы с помощью таких методов, как where или filter.

ActiveRecord:

Impression.where(company_id: 5).count

Django:

Impression.objects.filter(company_id=5).count()

По сути, если результирующий код SQL, выполняемый в базе данных, содержит предложение WHERE company_id = :value для каждой таблицы (включая таблицы в запросах JOIN), citus распознаёт, что запрос должен быть направлен на один узел, и выполнит его там без изменений. Это гарантирует доступность всех функций SQL. В конце концов, узел представляет собой обычный сервер Postgres Pro.

Кроме того, чтобы упростить процедуру, можно использовать библиотеку activerecord-multi-tenant для Ruby on Rails или django-multitenant для Django, которая автоматически добавит эти фильтры для всех пользовательских запросов, даже самых сложных. Ознакомьтесь с руководствами по миграции для Ruby on Rails и Django.

Данное руководство не привязано к определённой платформе, поэтому некоторые возможности citus показаны с использованием языка SQL. На другом языке эти операторы могут выглядеть иначе.

Ниже показан пример простого запроса и изменения, выполненных для одного арендатора.

-- Кампании с наибольшим бюджетом

SELECT name, cost_model, state, monthly_budget
  FROM campaigns
 WHERE company_id = 5
 ORDER BY monthly_budget DESC
 LIMIT 10;

-- Удвоение бюджета

UPDATE campaigns
   SET monthly_budget = monthly_budget*2
 WHERE company_id = 5;

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

-- Транзакция по перераспределению бюджета кампании

BEGIN;

UPDATE campaigns
   SET monthly_budget = monthly_budget + 1000
 WHERE company_id = 5
   AND id = 40;

UPDATE campaigns
   SET monthly_budget = monthly_budget - 1000
 WHERE company_id = 5
   AND id = 41;

COMMIT;

В качестве ещё одного примера поддержки SQL показан запрос с агрегатными и оконными функциями, который работает в citus так же, как и в Postgres Pro. Запрос упорядочивает объявления в каждой кампании по количеству показов.

SELECT a.campaign_id,
       RANK() OVER (
         PARTITION BY a.campaign_id
         ORDER BY a.campaign_id, count(*) desc
       ), count(*) as n_impressions, a.id
  FROM ads as a
  JOIN impressions as i
    ON i.company_id = a.company_id
   AND i.ad_id      = a.id
 WHERE a.company_id = 5
GROUP BY a.campaign_id, a.id
ORDER BY a.campaign_id, n_impressions desc;

Когда запросы ограничены одним арендатором, команды INSERT, UPDATE, DELETE, сложные команды SQL и транзакции работают как обычно.

До сих пор все таблицы распределялись по company_id, но существуют данные, которые могут использоваться всеми арендаторами и не «принадлежат» какому-либо конкретному арендатору. Например, все предприятия, использующие рекламную платформу из примера, могут получить географическую информацию о своей аудитории на основе IP-адресов. В базе данных одного компьютера это можно сделать с помощью справочной таблицы с geo-IP, как показано ниже. (Обратите внимание, что в настоящей таблице, вероятно, будет использоваться PostGIS.)

CREATE TABLE geo_ips (
  addrs cidr NOT NULL PRIMARY KEY,
  latlon point NOT NULL
    CHECK (-90  <= latlon[0] AND latlon[0] <= 90 AND
           -180 <= latlon[1] AND latlon[1] <= 180)
);
CREATE INDEX ON geo_ips USING gist (addrs inet_ops);

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

-- Создание синхронизированных копий geo_ips на всех рабочих узлах

SELECT create_reference_table('geo_ips');

Таблицы-справочники реплицируются на все рабочие узлы, и citus автоматически синхронизирует их при изменении. Обратите внимание, что вызывается функция create_reference_table, а не create_distributed_table.

После указания geo_ips в качестве справочной таблицы, загрузите в неё пример данных:

\copy geo_ips from 'geo_ips.csv' with csv

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

SELECT c.id, clicked_at, latlon
  FROM geo_ips, clicks c
 WHERE addrs >> c.user_ip
   AND c.company_id = 5
   AND c.ad_id = 290;

Ещё одна сложность при работе с многоарендными системами — синхронизация схем для всех арендаторов. Любое изменение схемы должно согласованно отражаться у всех арендаторов. В citus можно просто использовать стандартные DDL-команды Postgres Pro для изменения схемы таблиц, и расширение будет транслировать их с узла координатора на рабочие узлы, используя протокол двухфазной фиксации.

Например, в рекламных объявлениях в этом приложении могут использоваться сопроводительные надписи. Можно добавить столбец в таблицу, выполнив стандартный SQL-код на узле-координаторе:

ALTER TABLE ads
  ADD COLUMN caption text;

Это изменение попадёт на все рабочие узлы. После завершения этой команды кластер citus будет принимать запросы, которые читают или записывают данные в новый столбец caption.

За более подробным описанием трансляции DDL-команд по кластеру обратитесь к разделу Изменение таблиц.

Все арендаторы используют общую схему и аппаратную инфраструктуру, поэтому возникает вопрос: как быть с арендаторами, которые хотят хранить информацию, нужную только им? Например, один из арендаторов, использующих рекламную базу данных, может захотеть хранить в своём приложении информацию о cookie-файлах для отслеживания переходов, а другого арендатора может больше интересовать информация об агентах браузера. Традиционно базы данных, использующие подход с общей схемой для многоарендности, прибегали к созданию фиксированного числа предварительно выделенных «пользовательских» столбцов или к использованию внешних «таблиц расширения». Однако Postgres Pro предоставляет гораздо более простой способ благодаря своим неструктурированным типам столбцов, в частности, JSONB.

Обратите внимание, что в нашей схеме уже есть поле user_data типа JSONB, расположенное в clicks. Каждый арендатор может использовать его для гибкого хранения.

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

SELECT
  user_data->>'is_mobile' AS is_mobile,
  count(*) AS count
FROM clicks
WHERE company_id = 5
GROUP BY user_data->>'is_mobile'
ORDER BY count DESC;

Администратор базы данных даже может создать частичный индекс, чтобы повысить скорость обработки шаблонов запросов отдельного арендатора. Ниже представлен один из вариантов улучшения фильтров в компании с company_id = 5 для переходов от пользователей на мобильных устройствах:

CREATE INDEX click_user_data_is_mobile
ON clicks ((user_data->>'is_mobile'))
WHERE company_id = 5;

Кроме того, в Postgres Pro поддерживаются индексы GIN для JSONB. Создание индекса GIN для столбца JSONB приведёт к созданию индекса для каждого ключа и значения в этом документе JSON. Это ускоряет работу ряда операторов JSONB, таких как ?, ?| и ?&.

CREATE INDEX click_user_data
ON clicks USING gin (user_data);

-- эта часть ускоряет запросы "у каких переходов
-- в user_data есть ключ is_mobile?"

SELECT id
  FROM clicks
 WHERE user_data ? 'is_mobile'
   AND company_id = 5;

Многоарендные базы данных должны проектироваться с учётом масштабирования из-за роста бизнеса или желания арендаторов хранить больше данных. Для масштабирования citus не нужно вносить изменения или останавливать приложение — достаточно добавить новые устройства.

Возможность перебалансировки данных в кластере citus позволяет увеличивать размер данных или количество клиентов, а также повышать производительность по мере необходимости. Добавление новых компьютеров позволяет хранить данные в памяти, даже если их объём намного больше, чем ёмкость одного компьютера.

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

Чтобы масштабировать кластер citus, сначала добавьте в него новый рабочий узел с помощью функции citus_add_node.

Добавленный узел доступен в системе. Но сейчас на нём не хранятся никакие арендаторы, и citus пока не будет выполнять какие-либо запросы. Чтобы перенести существующие данные, укажите citus выполнить перебалансировку. Эта операция перемещает пакеты строк (сегменты) между активными узлами, чтобы сравнять их объёмы данных.

SELECT citus_rebalance_start();

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

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

Что касается первого вопроса, исследование данных крупных SaaS-сайтов показывает, что с ростом числа арендаторов размер данных арендаторов начинает следовать закону Ципфа. За подробностями обратитесь к рисунку ниже.

Например, в базе данных из 100 арендаторов прогнозируется, что на долю крупнейшего арендатора будет приходиться около 20% данных. В более реалистичном примере для крупной SaaS-компании с 10 000 арендаторов на долю крупнейшего из них будет приходиться около 2% данных. Даже при объёме данных 10 ТБ самому крупному арендатору потребуется 200 ГБ, для чего должно хватить всего одного узла.

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

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

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

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

SELECT isolate_tenant_to_new_shard(
  'companies', 5, 'CASCADE'
);

В результате будет получен идентификатор сегмента, выделенного для хранения company_id=5:

┌─────────────────────────────┐
│ isolate_tenant_to_new_shard │
├─────────────────────────────┤
│                      102240 │
└─────────────────────────────┘

Затем переместите данные по сети на новый выделенный узел. Создайте новый узел, как описано в предыдущем разделе. Обратите внимание на адрес узла.

-- Найдите узел, на котором сейчас находится новый сегмент

SELECT nodename, nodeport
  FROM pg_dist_placement AS placement,
       pg_dist_node AS node
 WHERE placement.groupid = node.groupid
   AND node.noderole = 'primary'
   AND shardid = 102240;

-- Переместите сегмент на выбранный рабочий узел (другие сегменты, созданные
-- с параметром CASCADE также будут перемещены)

-- Обратите внимание, что на для всех узлов следует задать для wal_level значение >= logical,
-- чтобы использовать citus_move_shard_placement.
-- Также необходимо перезапустить кластер после изменения wal_level в
-- файлах postgresql.conf.

SELECT citus_move_shard_placement(
  102240,
  'source_host', source_port,
  'dest_host', dest_port);

Подтвердите перемещение сегмента, повторно отправив запрос таблице pg_dist_placement.

Теперь вы знаете, как использовать citus для обеспечения масштабируемости многоарендного приложения. Если схема уже существует и её нужно перенести для citus, см. раздел Миграция существующего приложения.

Чтобы настроить клиентское приложение, например Ruby on Rails или Django, используйте руководство по миграции Ruby on Rails или Django.

Данные в примере представляют собой неизменяемый поток данных журнала. Они будут вставляться непосредственно в citus, но часто эти данные сначала перенаправляются через платформу Kafka или её аналог. Это упрощает предварительное агрегирование, когда объёмы данных становятся неконтролируемо большими.

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

-- Выполните на узле-координаторе

CREATE TABLE http_request (
  site_id INT,
  ingest_time TIMESTAMPTZ DEFAULT now(),

  url TEXT,
  request_country TEXT,
  ip_address TEXT,

  status_code INT,
  response_time_msec INT
);

SELECT create_distributed_table('http_request', 'site_id');

Когда вызывается функция create_distributed_table, citus получает указание распределить по хешу http_request, используя столбец site_id. Это означает, что все данные конкретного сайта будут храниться в одном сегменте.

Пользовательские функции используют конфигурацию со стандартным количеством сегментов. Рекомендуется использовать в 2–4 раза больше сегментов, чем количество ядер ЦП в кластере. Такое количество сегментов позволяет перебалансировать данные в кластере после добавления новых рабочих узлов.

Теперь система готова принимать данные и обслуживать запросы. Запустите следующий цикл в консоли psql в фоновом режиме, продолжая выполнять другие команды из этой статьи. Этот цикл генерирует искусственные данные каждые 1-2 секунды.

DO $$
  BEGIN LOOP
    INSERT INTO http_request (
      site_id, ingest_time, url, request_country,
      ip_address, status_code, response_time_msec
    ) VALUES (
      trunc(random()*32), clock_timestamp(),
      concat('http://example.com/', md5(random()::text)),
      ('{China,India,USA,Indonesia}'::text[])[ceil(random()*4)],
      concat(
        trunc(random()*250 + 2), '.',
        trunc(random()*250 + 2), '.',
        trunc(random()*250 + 2), '.',
        trunc(random()*250 + 2)
      )::inet,
      ('{200,404}'::int[])[ceil(random()*2)],
      5+trunc(random()*150)
    );
    COMMIT;
    PERFORM pg_sleep(random() * 0.25);
  END LOOP;
END $$;

После наполнения данными можно запускать запросы для панели, например:

SELECT
  site_id,
  date_trunc('minute', ingest_time) as minute,
  COUNT(1) AS request_count,
  SUM(CASE WHEN (status_code between 200 and 299) THEN 1 ELSE 0 END) as success_count,
  SUM(CASE WHEN (status_code between 200 and 299) THEN 0 ELSE 1 END) as error_count,
  SUM(response_time_msec) / COUNT(1) AS average_response_time_msec
FROM http_request
WHERE date_trunc('minute', ingest_time) > now() - '5 minutes'::interval
GROUP BY site_id, minute
ORDER BY minute ASC;

Описанный выше вариант работает, но у него есть два недостатка:

Оба недостатка можно преодолеть путём группирования данных в предварительно агрегированную форму. Здесь необработанные данные группируются в таблицу, в которой хранятся сводные данные по минутным интервалам. В производственной системе, вероятно, также понадобятся интервалы в 1 час и 1 день, каждый из которых соответствует уровням масштабирования на информационной панели. Если пользователю нужны данные за последний месяц, панель может просто прочитать и составить график значений за каждый из последних 30 дней.

CREATE TABLE http_request_1min (
  site_id INT,
  ingest_time TIMESTAMPTZ, -- минута, представленная данной строкой

  error_count INT,
  success_count INT,
  request_count INT,
  average_response_time_msec INT,
  CHECK (request_count = error_count + success_count),
  CHECK (ingest_time = date_trunc('minute', ingest_time))
);

SELECT create_distributed_table('http_request_1min', 'site_id');

CREATE INDEX http_request_1min_idx ON http_request_1min (site_id, ingest_time);

Этот блок кода очень похож на предыдущий. Самое главное: в нём также сегментируется site_id и используется та же стандартная конфигурация для количества сегментов. Поскольку все три этих элемента совпадают, сегменты http_request и сегменты http_request_1min полностью соответствуют друг другу, и citus разместит соответствующие сегменты на одном рабочем узле. Этот процесс называется совмещением. Он ускоряет выполнение соединений и позволяет использовать свёртки. За более подробным описанием обратитесь к рисунку.

Чтобы заполнить http_request_1min, нужно периодически запускать команду INSERT INTO SELECT. Это возможно, поскольку таблицы совмещены. Следующая функция упрощает свёртку.

-- Однострочная таблица для хранения момента последней свёртки
CREATE TABLE latest_rollup (
  minute timestamptz PRIMARY KEY,

  -- Параметр "minute" должен усекать с точностью до минут
  CHECK (minute = date_trunc('minute', minute))
);

-- Инициализация по очень старой дате
INSERT INTO latest_rollup VALUES ('10-10-1901');

-- Функция для выполнения группирования
CREATE OR REPLACE FUNCTION rollup_http_request() RETURNS void AS $$
DECLARE
  curr_rollup_time timestamptz := date_trunc('minute', now() - interval '1 minute');
  last_rollup_time timestamptz := minute from latest_rollup;
BEGIN
  INSERT INTO http_request_1min (
    site_id, ingest_time, request_count,
    success_count, error_count, average_response_time_msec
  ) SELECT
    site_id,
    date_trunc('minute', ingest_time),
    COUNT(1) as request_count,
    SUM(CASE WHEN (status_code between 200 and 299) THEN 1 ELSE 0 END) as success_count,
    SUM(CASE WHEN (status_code between 200 and 299) THEN 0 ELSE 1 END) as error_count,
    SUM(response_time_msec) / COUNT(1) AS average_response_time_msec
  FROM http_request
  -- roll up only data new since last_rollup_time
  WHERE date_trunc('minute', ingest_time) <@
          tstzrange(last_rollup_time, curr_rollup_time, '(]')
  GROUP BY 1, 2;

  -- Обновление значения в latest_rollup, чтобы при следующем запуске
  -- свёртки она применялась к данным, новее, чем curr_rollup_time
  UPDATE latest_rollup SET minute = curr_rollup_time;
END;
$$ LANGUAGE plpgsql;

* * * * * psql -c 'SELECT rollup_http_request();'

В качестве альтернативы можно использовать приложение pg_cron (или его аналог), позволяющее составлять расписание повторяющихся запросов прямо из БД.

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

SELECT site_id, ingest_time as minute, request_count,
       success_count, error_count, average_response_time_msec
  FROM http_request_1min
 WHERE ingest_time > date_trunc('minute', now()) - '5 minutes'::interval;

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

DELETE FROM http_request WHERE ingest_time < now() - interval '1 day';
DELETE FROM http_request_1min WHERE ingest_time < now() - interval '1 month';

В производственной среде можно обернуть эти запросы в функцию и вызывать её каждую минуту в задании cron.

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

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

В следующих разделах более подробно рассматривается базовая архитектура и решение часто возникающих вопросов.

Распространённый вопрос в HTTP-аналитике связан с приблизительным подсчётом уникальных значений: например сколько уникальных посетителей было на сайте за последний месяц. Чтобы ответить на этот вопрос точно, необходимо хранить в таблицах свёрток список всех замеченных посетителей, а это слишком большой объём данных. Использование приблизительного количества требует гораздо меньших ресурсов.

Тип данных, называемый HyperLogLog или hll, может дать приблизительный ответ на запрос. При этом, чтобы определить приблизительное количество уникальных элементов в наборе требуется совсем немного места. Точность определения можно регулировать. Используя всего 1280 байт можно учесть до десятков миллиардов уникальных посетителей с погрешностью не более 2,2%.

Аналогичная проблема возникает при необходимости выполнить глобальный запрос, например, узнать количество уникальных IP-адресов, посетивших любой из сайтов клиента за последний месяц. Без hll этот запрос предполагает отправку списков IP-адресов с рабочих узлов координатору для исключения дубликатов. Это одновременно и большого сетевой трафик, и большой объём вычислений. При использовании hll можно значительно повысить скорость запросов.

Установите расширение hll, инструкции для которого можно найти в репозитории GitHub, и включите его:

CREATE EXTENSION hll;

Теперь можно отслеживать IP-адреса в свёртке с помощью расширения hll. Сначала добавьте столбец в таблицу свёрток.

ALTER TABLE http_request_1min ADD COLUMN distinct_ip_addresses hll;

Затем используйте пользовательскую агрегацию для заполнения столбца. Для этого добавьте её в запрос в функции свёртки:

@@ -1,10 +1,12 @@
  INSERT INTO http_request_1min (
    site_id, ingest_time, request_count,
    success_count, error_count, average_response_time_msec
+   , distinct_ip_addresses
  ) SELECT
    site_id,
    date_trunc('minute', ingest_time),
    COUNT(1) as request_count,
    SUM(CASE WHEN (status_code between 200 and 299) THEN 1 ELSE 0 END) as success_count,
    SUM(CASE WHEN (status_code between 200 and 299) THEN 0 ELSE 1 END) as error_count,
    SUM(response_time_msec) / COUNT(1) AS average_response_time_msec
+   , hll_add_agg(hll_hash_text(ip_address)) AS distinct_ip_addresses
  FROM http_request

Запросы информационной панели немного сложнее: нужно считывать определённое количество IP-адресов, вызывая функцию hll_cardinality:

SELECT site_id, ingest_time as minute, request_count,
       success_count, error_count, average_response_time_msec,
       hll_cardinality(distinct_ip_addresses) AS distinct_ip_address_count
  FROM http_request_1min
 WHERE ingest_time > date_trunc('minute', now()) - interval '5 minutes';

Тип данных hll не просто позволяет ускорить работу, но и добавляет ранее недоступные возможности. Допустим, были выполнены свёртки, но вместо использования hll сохранилось точное количество уникальных значений. Такой способ тоже работает, но при этом невозможно ответить на такие запросы, как «сколько отдельных сеансов, для которых были удалены необработанные данные, было в течение этого недельного периода в прошлом?».

С hll ответить на такой запрос легко. Можно вычислить количество различных IP-адресов за определённый период времени с помощью следующего запроса:

SELECT hll_cardinality(hll_union_agg(distinct_ip_addresses))
FROM http_request_1min
WHERE ingest_time > date_trunc('minute', now()) - '5 minutes'::interval;

Чтобы узнать больше о расширении hll, обратитесь к документации в репозитории проекта.

Расширение citus совместимо со встроенной поддержкой Postgres Pro для неструктурированных типов данных. В качестве примера можно вычислить количество посетителей из каждой страны. Использование полуструктурированного типа данных избавляет от необходимости добавлять столбец для каждой отдельной страны и в конечном итоге получать строки с сотнями полупустых столбцов. Рекомендуется использовать формат JSONB, и в примере будет показано, как включить столбцы JSONB в модель данных.

Сначала добавьте новый столбец в таблицу свёрток:

ALTER TABLE http_request_1min ADD COLUMN country_counters JSONB;

Затем включите его в свёртки, изменив соответствующую функцию:

@@ -1,14 +1,19 @@
  INSERT INTO http_request_1min (
    site_id, ingest_time, request_count,
    success_count, error_count, average_response_time_msec
+   , country_counters
  ) SELECT
    site_id,
    date_trunc('minute', ingest_time),
    COUNT(1) as request_count,
    SUM(CASE WHEN (status_code between 200 and 299) THEN 1 ELSE 0 END) as success_count
    SUM(CASE WHEN (status_code between 200 and 299) THEN 0 ELSE 1 END) as error_count
    SUM(response_time_msec) / COUNT(1) AS average_response_time_msec
- FROM http_request
+   , jsonb_object_agg(request_country, country_count) AS country_counters
+ FROM (
+   SELECT *,
+     count(1) OVER (
+       PARTITION BY site_id, date_trunc('minute', ingest_time), request_country
+     ) AS country_count
+   FROM http_request
+ ) h

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

SELECT
  request_count, success_count, error_count, average_response_time_msec,
  COALESCE(country_counters->>'USA', '0')::int AS american_visitors
FROM http_request_1min
WHERE ingest_time > date_trunc('minute', now()) - '5 minutes'::interval;

Можно объединить методы секционирования таблиц с одним узлом с распределённым сегментированием citus, чтобы создать масштабируемую базу данных временных рядов. Такой подход позволит использовать преимущества обоих методов. Особенно элегантно такое решение смотрится при декларативном секционировании таблиц Postgres Pro. За подробностями обратитесь к рисунку.

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

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

Первым шагом будет создание и секционирование таблицы по времени в базе данных Postgres Pro на одном узле:

-- Декларативно секционированные таблицы
CREATE TABLE github_events (
  event_id bigint,
  event_type text,
  event_public boolean,
  repo_id bigint,
  payload jsonb,
  repo jsonb,
  actor jsonb,
  org jsonb,
  created_at timestamp
) PARTITION BY RANGE (created_at);

Обратите внимание на PARTITION BY RANGE (create_at). Это указание для Postgres Pro, что таблица будет секционирована по столбцу created_at в упорядоченных диапазонах. Но никакие секции для конкретных диапазонов ещё не были созданы.

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

SELECT create_distributed_table('github_events', 'repo_id');

На данном этапе расширением citus были созданы сегменты для этой таблицы на рабочих узлах. Внутри каждый сегмент представляет собой таблицу с именем github_events_N для каждого идентификатора сегмента N. Кроме того, была распространена информация о секционировании, и в каждом из упомянутых сегментов был объявлен ключ секционирования Partition key: RANGE (created_at).

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

Расширение citus предоставляет вспомогательные функции для управления секциями. Можно создать порцию ежемесячных секций, используя функцию create_time_partitions:

SELECT create_time_partitions(
  table_name         := 'github_events',
  partition_interval := '1 month',
  end_at             := now() + '12 months'
);

В citus также есть представление time_partitions, позволяющее легко исследовать секции, созданные расширением.

SELECT partition
  FROM time_partitions
 WHERE parent_table = 'github_events'::regclass;

┌────────────────────────┐
│       partition        │
├────────────────────────┤
│ github_events_p2021_10 │
│ github_events_p2021_11 │
│ github_events_p2021_12 │
│ github_events_p2022_01 │
│ github_events_p2022_02 │
│ github_events_p2022_03 │
│ github_events_p2022_04 │
│ github_events_p2022_05 │
│ github_events_p2022_06 │
│ github_events_p2022_07 │
│ github_events_p2022_08 │
│ github_events_p2022_09 │
│ github_events_p2022_10 │
└────────────────────────┘

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

-- Установите запуск двух заданий cron раз в месяц:

-- 1. Обеспечьте наличие секций на протяжении следующих 12 месяцев

SELECT cron.schedule('create-partitions', '0 0 1 * *', $$
  SELECT create_time_partitions(
      table_name         := 'github_events',
      partition_interval := '1 month',
      end_at             := now() + '12 months'
  )
$$);

-- 2. (Необязательно) Обеспечьте хранение данных не более, чем за год

SELECT cron.schedule('drop-partitions', '0 0 1 * *', $$
  CALL drop_old_time_partitions(
      'github_events',
      now() - interval '12 months' /* older_than */
  );
$$);

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

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

Вновь рассмотрим пример с событиями GitHub. Создайте новую таблицу с именем github_columnar_events во избежание путаницы с предыдущим примером. Чтобы полностью сосредоточиться на аспекте столбцового хранения, данная таблица не будет распределяться.

Затем загрузите пример с данными:

wget http://examples.citusdata.com/github_archive/github_events-2015-01-01-{0..5}.csv.gz
gzip -c -d github_events-2015-01-01-*.gz >> github_events.csv

-- Новая таблица с такой же структурой, как в примере
-- из предыдущего раздела

CREATE TABLE github_columnar_events ( LIKE github_events )
PARTITION BY RANGE (created_at);

-- Создайте секции для хранения данных за два часа в каждой

SELECT create_time_partitions(
  table_name         := 'github_columnar_events',
  partition_interval := '2 hours',
  start_from         := '2015-01-01 00:00:00',
  end_at             := '2015-01-01 08:00:00'
);

-- Заполните таблицы загруженными данными
-- (обратите внимание, что для этих данных требуется, чтобы БД имела кодировку UTF8)

\COPY github_columnar_events FROM 'github_events.csv' WITH (format CSV)

-- Выведите список секций, чтобы убедиться, что
-- используется строковое хранение (метод доступа heap)

SELECT partition, access_method
  FROM time_partitions
 WHERE parent_table = 'github_columnar_events'::regclass;

┌─────────────────────────────────────────┬───────────────┐
│                partition                │ access_method │
├─────────────────────────────────────────┼───────────────┤
│ github_columnar_events_p2015_01_01_0000 │ heap          │
│ github_columnar_events_p2015_01_01_0200 │ heap          │
│ github_columnar_events_p2015_01_01_0400 │ heap          │
│ github_columnar_events_p2015_01_01_0600 │ heap          │
└─────────────────────────────────────────┴───────────────┘

-- Преобразуйте старые секции для столбцового хранения

CALL alter_old_partitions_set_access_method(
  'github_columnar_events',
  '2015-01-01 06:00:00' /* older_than */,
  'columnar'
);

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

SELECT partition, access_method
  FROM time_partitions
 WHERE parent_table = 'github_columnar_events'::regclass;

┌─────────────────────────────────────────┬───────────────┐
│                partition                │ access_method │
├─────────────────────────────────────────┼───────────────┤
│ github_columnar_events_p2015_01_01_0000 │ columnar      │
│ github_columnar_events_p2015_01_01_0200 │ columnar      │
│ github_columnar_events_p2015_01_01_0400 │ columnar      │
│ github_columnar_events_p2015_01_01_0600 │ heap          │
└─────────────────────────────────────────┴───────────────┘

Чтобы увидеть степень сжатия столбцовой таблицы, используйте VACUUM VERBOSE. Степень сжатия для трёх столбцовых секций выглядит так:

VACUUM VERBOSE github_columnar_events;

INFO:  statistics for "github_columnar_events_p2015_01_01_0000":
storage id: 10000000003
total file size: 4481024, total data size: 4444425
compression rate: 8.31x
total row count: 15129, stripe count: 1, average rows per stripe: 15129
chunk count: 18, containing data for dropped columns: 0, zstd compressed: 18

INFO:  statistics for "github_columnar_events_p2015_01_01_0200":
storage id: 10000000004
total file size: 3579904, total data size: 3548221
compression rate: 8.26x
total row count: 12714, stripe count: 1, average rows per stripe: 12714
chunk count: 18, containing data for dropped columns: 0, zstd compressed: 18

INFO:  statistics for "github_columnar_events_p2015_01_01_0400":
storage id: 10000000005
total file size: 2949120, total data size: 2917407
compression rate: 8.51x
total row count: 11756, stripe count: 1, average rows per stripe: 11756
chunk count: 18, containing data for dropped columns: 0, zstd compressed: 18

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

SELECT COUNT(DISTINCT repo_id)
  FROM github_columnar_events;

┌───────┐
│ count │
├───────┤
│ 16001 │
└───────┘

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

-- Ежемесячное задание cron

SELECT cron.schedule('compress-partitions', '0 0 1 * *', $$
  CALL alter_old_partitions_set_access_method(
    'github_columnar_events',
    now() - interval '6 months' /* older_than */,
    'columnar'
  );
$$);

Традиционный способ сегментирования таблиц citus — это модель общей схемы в единой базе данных, также известная как сегментирование на основе строк, в которой арендаторы сосуществуют как строки одной таблицы. Арендатор определяется путём установления столбца распределения, который позволяет разделить таблицу горизонтально.

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

Преимущества:

Недостатки:

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

Преимущества:

Недостатки:

В кластере citus имеется несколько типов таблиц, каждый из которых используется для разных целей.

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

Таблица метаданных pg_dist_shard на узле-координаторе содержит одну строку для каждого сегмента каждой распределённой таблицы в системе. Строка соответствует shardid с диапазоном целых чисел в хеш-пространстве (shardminvalue, shardmaxvalue):

SELECT * FROM pg_dist_shard;
 logicalrelid  | shardid | shardstorage | shardminvalue | shardmaxvalue
---------------+---------+--------------+---------------+---------------
 github_events |  102026 | t            | 268435456     | 402653183
 github_events |  102027 | t            | 402653184     | 536870911
 github_events |  102028 | t            | 536870912     | 671088639
 github_events |  102029 | t            | 671088640     | 805306367
 (4 rows)

Если узлу-координатору нужно определить, какой сегмент содержит строку github_events, значение столбца распределения в строке хешируется и проверяется, какой диапазон сегмента содержит хешированное значение. (Диапазоны определены так, что образ хеш-функции является их несвязным объединением.)

SELECT
    shardid,
    node.nodename,
    node.nodeport
FROM pg_dist_placement placement
JOIN pg_dist_node node
  ON placement.groupid = node.groupid
 AND node.noderole = 'primary'::noderole
WHERE shardid = 102027;

┌─────────┬───────────┬──────────┐
│ shardid │ nodename  │ nodeport │
├─────────┼───────────┼──────────┤
│  102027 │ localhost │     5433 │
└─────────┴───────────┴──────────┘

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

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

За подробным описанием и примерами обратитесь к разделу Совмещение таблиц.

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

Запросы, читающие или влияющие на сегменты, которые равномерно распределены по множеству узлов, могут выполняться в «реальном времени». Обратите внимание, что результаты запроса должны передаваться обратно через узел-координатор, поэтому ускорение работы наиболее очевидно при составных окончательных результатах, например, при использовании агрегатных функций, таких как подсчёт количества значений и описательная статистика.

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

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

На следующей диаграмме показано совмещение в многоарендной модели данных. Она содержит две таблицы: «Accounts» (Учётные записи) и «Campaigns» (Кампании), каждая из которых распределена по account_id. Сегменты обозначены как прямоугольники, цвет которых соответствует цвету содержащего их рабочего узла. Зелёные сегменты хранятся вместе на одном рабочем узле, а синие — на другом. Обратите внимание, что запрос соединения учётных записей и кампаний будет содержать все необходимые данные на одном узле с выборкой по одному account_id.

Чтобы применить этот подход к собственной схеме, сначала определите, кто в приложении будет арендатором. Обычно это предприятие, учётная запись, организация или клиент. При этом имя столбца будет примерно таким: company_id или customer_id. Следует проанализировать каждый запрос и ответить на вопрос: сработает ли такой запрос с дополнительным предложением WHERE, ограничивающим все задействованные таблицы строками с одним и тем же tenant_id? Запросы в многоарендной модели обычно ограничиваются арендатором, например, запросы о продажах или наличии товара будут ограничены определённым магазином.

Практические рекомендации:

Подробный пример создания такого типа приложений описан в разделе Многоарендные приложения.

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

В качестве столбца распределения в многоарендной модели используется идентификатор арендатора, а в модели реального времени — «идентификатор объекта». Типичными объектами являются пользователи, узлы или устройства.

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

Практические рекомендации:

Подробный пример создания такого типа приложений описан в разделе Панели для анализа в реальном времени.

При использовании временных рядов приложения запрашивают последнюю информацию, одновременно архивируя старую информацию.

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

Практические рекомендации:

Подробный пример создания такого типа приложений описан в разделе Данные временных рядов.

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

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

CREATE TABLE event (
  tenant_id int,
  event_id bigint,
  page_id int,
  payload jsonb,
  primary key (tenant_id, event_id)
);

CREATE TABLE page (
  tenant_id int,
  page_id int,
  path text,
  primary key (tenant_id, page_id)
);

SELECT page_id, count(event_id)
FROM
  page
LEFT JOIN  (
  SELECT * FROM event
  WHERE (payload->>'time')::timestamptz >= now() - interval '1 week'
) recent
USING (tenant_id, page_id)
WHERE tenant_id = 6 AND path LIKE '/blog%'
GROUP BY page_id;

-- По неопытности для примера используем event_id и page_id в качестве столбцов распределения

SELECT create_distributed_table('event', 'event_id');
SELECT create_distributed_table('page', 'page_id');

SELECT page_id FROM page WHERE path LIKE '/blog%' AND tenant_id = 6;

SELECT page_id, count(*) AS count
FROM event
WHERE page_id IN (/*…page IDs from first query…*/)
  AND tenant_id = 6
  AND (payload->>'time')::date >= now() - interval '1 week'
GROUP BY page_id ORDER BY count DESC LIMIT 10;

-- Совмещение таблиц с помощью общего столбца распределения
SELECT create_distributed_table('event', 'tenant_id');
SELECT create_distributed_table('page', 'tenant_id', colocate_with => 'event');

SELECT page_id, count(event_id)
FROM
  page
LEFT JOIN  (
  SELECT * FROM event
  WHERE (payload->>'time')::timestamptz >= now() - interval '1 week'
) recent
USING (tenant_id, page_id)
WHERE tenant_id = 6 AND path LIKE '/blog%'
GROUP BY page_id;

После определения объёма необходимых изменений БД следующим важным шагом является изменение структуры данных существующей БД. Сначала в таблицы, требующие заполнения, добавляется столбец ключа распределения.

-- Денормализация таблицы line_items путём добавления столбца store_id

ALTER TABLE line_items ADD COLUMN store_id uuid;

UPDATE line_items
   SET store_id = orders.store_id
  FROM line_items
 INNER JOIN orders
 WHERE line_items.order_id = orders.order_id;

-- Функция для заполнения до тысячи
-- строк в таблице line_items

CREATE FUNCTION backfill_batch()
RETURNS void LANGUAGE sql AS $$
  WITH batch AS (
    SELECT line_items_id, order_id
      FROM line_items
     WHERE store_id IS NULL
     LIMIT 1000
       FOR UPDATE
      SKIP LOCKED
  )
  UPDATE line_items AS li
     SET store_id = orders.store_id
    FROM batch, orders
   WHERE batch.line_item_id = li.line_item_id
     AND batch.order_id = orders.order_id;
$$;

-- Функция выполняется каждые 15 минут
SELECT cron.schedule('*/15 * * * *', 'SELECT backfill_batch()');

-- Обратите внимание на возвращаемое значение cron.schedule

-- Предполагается, что 42 — полученный идентификатор
-- задания cron.schedule

SELECT cron.unschedule(42);

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

Такие приложения, как конвейеры данных о событиях и информационные панели для анализа в реальном времени, требуют выполнения запросов за доли секунды на больших объёмах данных. Один из способов ускорить эти запросы — заранее вычислить и сохранить агрегированные данные. Это называется «сворачиванием» данных и позволяет избежать затрат на обработку первичных данных во время выполнения запроса. Ещё одно преимущество сворачивания данных временных рядов в почасовую или ежедневную статистику — экономия места. Старые данные могут удаляться, когда больше нет необходимости хранить все подробности и достаточно агрегированных данных.

В примере рассматривается распределённая таблица для отслеживания просмотров страниц по URL:

CREATE TABLE page_views (
  site_id int,
  url text,
  host_ip inet,
  view_time timestamp default now(),

  PRIMARY KEY (site_id, url)
);

SELECT create_distributed_table('page_views', 'site_id');

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

-- Количество просмотров по каждому URL-адресу за день на сайте 5
SELECT view_time::date AS day, site_id, url, count(*) AS view_count
  FROM page_views
  WHERE site_id = 5 AND
    view_time >= date '2016-01-01' AND view_time < date '2017-01-01'
  GROUP BY view_time::date, site_id, url;

Описанная выше конфигурация работает, но имеет два недостатка. Во-первых, при многократном выполнении запроса агрегирования должны каждый раз прочитываться все связанные строки и вычисляться результаты для всего набора данных. Если этот запрос используется для отображения информационной панели, быстрее сохранить агрегированные результаты в таблице с ежедневными просмотрами страниц и запрашивать эту таблицу. Во-вторых, стоимость хранения будет расти пропорционально объёму данных и длине периода, за который доступны запросы. На практике лучше сохранять необработанные события в течение короткого периода времени, а для долгосрочного анализа использовать исторические графики.

Для хранения ежедневной статистики можно создать таблицу daily_page_views.

CREATE TABLE daily_page_views (
  site_id int,
  day date,
  url text,
  view_count bigint,
  PRIMARY KEY (site_id, day, url)
);

SELECT create_distributed_table('daily_page_views', 'site_id');

В этом примере таблицы page_views и daily_page_views распределяются по столбцу site_id. При этом гарантируется, что данные, соответствующие конкретному сайту, будут совмещены на одном узле. Сохранение строк двух таблиц вместе на каждом узле минимизирует сетевой трафик между узлами и обеспечивает выполнение с высокой степенью распараллеливания.

После создания новой распределённой таблицы можно выполнить INSERT INTO ... SELECT, чтобы объединить необработанные представления страниц в агрегированную таблицу. Далее агрегируются данные о просмотрах страниц каждый день. Пользователи citus часто ждут некоторое время после окончания дня, чтобы выполнить подобный запрос и учесть поздно поступающие данные.

-- Группирование вчерашних данных
INSERT INTO daily_page_views (day, site_id, url, view_count)
  SELECT view_time::date AS day, site_id, url, count(*) AS view_count
  FROM page_views
  WHERE view_time >= date '2017-01-01' AND view_time < date '2017-01-02'
  GROUP BY view_time::date, site_id, url;

-- Теперь результаты доступны прямо из таблицы
SELECT day, site_id, url, view_count
  FROM daily_page_views
  WHERE site_id = 5 AND
    day >= date '2016-01-01' AND day < date '2017-01-01';

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

Ситуация меняется, когда приходится иметь дело с данными, поступающими с задержкой, или выполнять запрос со свёрткой более одного раза в день. Если новые строки уже соответствуют дням в таблице свёртки, количество совпадений должно увеличиться. Postgres Pro может справиться с этой ситуацией с помощью ON CONFLICT — встроенным методом выполнения UPSERTS. Ниже представлен пример этого метода.

-- Группирование, начиная с указанной даты,
-- ежедневные просмотры страницы изменяются при необходимости
INSERT INTO daily_page_views (day, site_id, url, view_count)
  SELECT view_time::date AS day, site_id, url, count(*) AS view_count
  FROM page_views
  WHERE view_time >= date '2017-01-01'
  GROUP BY view_time::date, site_id, url
  ON CONFLICT (day, url, site_id) DO UPDATE SET
    view_count = daily_page_views.view_count + EXCLUDED.view_count;

DELETE FROM github_events
WHERE repo_id IN (24509048, 24509049);

UPDATE github_events
SET event_public = TRUE
WHERE (org->>'id')::int = 5430905;

SET citus.multi_shard_commit_protocol = '2pc';

-- Поскольку таблица github_events распределена по столбцу repo_id,
-- операция будет выполняться на одном рабочем узле

DELETE FROM github_events
WHERE repo_id = 206084;

BEGIN;

  -- Обращение к событиям по repo_id, но
  -- с их блокировкой для записи
  SELECT *
  FROM github_events
  WHERE repo_id = 206084
  FOR UPDATE;

  -- Вычисление нужного значения event_public с использованием
  -- логики приложения, использующей строки...

  -- Применение изменений
  UPDATE github_events
  SET event_public = :our_new_value
  WHERE repo_id = 206084;

COMMIT;

Как обсуждалось в предыдущих разделах, citus расширяет возможности Postgres Pro для выполнения распределённых вычислений. Это означает, что можно использовать стандартные запросы Postgres Pro SELECT на координаторе citus. Затем расширение распараллеливает запросы SELECT, включающие сложные выборки, группировки и упорядочивания, а также JOIN, чтобы ускорить их выполнение. На высоком уровне citus разбивает запрос SELECT на более мелкие фрагменты, назначает их рабочим узлам, контролирует их выполнение, объединяет результаты (и сортирует их, если необходимо) и возвращает итоговый результат пользователю.

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

SELECT sum(value1), sum(distinct value2) FROM distributed_table;

SET citus.coordinator_aggregation_strategy TO 'disabled';

-- Несколько различных функций count в одном запросе обычно выполняются медленно
SELECT count(distinct a), count(distinct b), count(distinct c)
FROM table_abc;

-- Сначала ничего нет, фиксируем, что увидели букву «а»
SELECT topn_add('{}', 'a');
-- => {"a": 1}

-- Фиксируем появление ещё одной буквы «а»
SELECT topn_add(topn_add('{}', 'a'), 'a');
-- => {"a": 2}

-- Для normal_rand
CREATE EXTENSION tablefunc;

-- Подсчёт значений из нормального распределения
SELECT topn_add_agg(floor(abs(i))::text)
  FROM normal_rand(1000, 5, 0.7) i;
-- => {"2": 1, "3": 74, "4": 420, "5": 425, "6": 77, "7": 3}

curl -L https://examples.citusdata.com/customer_reviews_2000.csv.gz | \
  gunzip > reviews.csv

CREATE TABLE customer_reviews
(
    customer_id TEXT,
    review_date DATE,
    review_rating INTEGER,
    review_votes INTEGER,
    review_helpful_votes INTEGER,
    product_id CHAR(10),
    product_title TEXT,
    product_sales_rank BIGINT,
    product_group TEXT,
    product_category TEXT,
    product_subcategory TEXT,
    similar_product_ids CHAR(10)[]
);

SELECT create_distributed_table('customer_reviews', 'product_id');

\COPY customer_reviews FROM 'reviews.csv' WITH CSV

-- Выполните указанную команду на узле-кординаторе, она будет транслирована и на рабочие узлы
CREATE EXTENSION topn;

-- Таблица для материализации ежедневно агрегируемых данных
CREATE TABLE reviews_by_day
(
  review_date date unique,
  agg_data jsonb
);

SELECT create_reference_table('reviews_by_day');

-- Материализовать количество отзывов по каждому продукту за день для каждого покупателя
INSERT INTO reviews_by_day
  SELECT review_date, topn_add_agg(product_id)
  FROM customer_reviews
  GROUP BY review_date;

SELECT review_date, (topn(agg_data, 1)).*
FROM reviews_by_day
ORDER BY review_date
LIMIT 5;

┌─────────────┬────────────┬───────────┐
│ review_date │    item    │ frequency │
├─────────────┼────────────┼───────────┤
│ 2000-01-01  │ 0939173344 │        12 │
│ 2000-01-02  │ B000050XY8 │        11 │
│ 2000-01-03  │ 0375404368 │        12 │
│ 2000-01-04  │ 0375408738 │        14 │
│ 2000-01-05  │ B00000J7J4 │        17 │
└─────────────┴────────────┴───────────┘

SELECT (topn(topn_union_agg(agg_data), 5)).*
FROM reviews_by_day
WHERE review_date >= '2000-01-01' AND review_date < '2000-02-01'
ORDER BY 2 DESC;

┌────────────┬───────────┐
│    item    │ frequency │
├────────────┼───────────┤
│ 0375404368 │       217 │
│ 0345417623 │       217 │
│ 0375404376 │       217 │
│ 0375408738 │       217 │
│ 043936213X │       204 │
└────────────┴───────────┘

SET citus.limit_clause_row_fetch_count TO 10000;

Кластер citus состоит из экземпляра узла-координатора и нескольких экземпляров рабочих узлов. Данные сегментируются по рабочим узлам, а на узле-координаторе хранятся метаданные об этих сегментах. Все запросы к кластеру выполняются через узел-координатор. Он разбивает запрос на более мелкие фрагменты, где каждый фрагмент может выполняться независимо в отдельном сегменте. Затем узел-координатор назначает фрагменты рабочим узлам, контролирует их выполнение, объединяет их результаты и возвращает конечный результат пользователю. Краткое описание архитектуры обработки запросов представлено на диаграмме ниже.

Конвейер обработки запросов citus содержит два компонента:

Они описаны более подробно в следующих разделах.

SELECT page_id, count(distinct host_ip)
FROM page_views
WHERE page_id IN (
  SELECT page_id
  FROM page_views
  GROUP BY page_id
  ORDER BY count(*) DESC
  LIMIT 20
)
GROUP BY page_id;

GroupAggregate  (cost=0.00..0.00 rows=0 width=0)
  Group Key: remote_scan.page_id
  ->  Sort  (cost=0.00..0.00 rows=0 width=0)
    Sort Key: remote_scan.page_id
    ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)
      ->  Distributed Subplan 6_1
        ->  Limit  (cost=0.00..0.00 rows=0 width=0)
          ->  Sort  (cost=0.00..0.00 rows=0 width=0)
            Sort Key: COALESCE((pg_catalog.sum((COALESCE((pg_catalog.sum(remote_scan.worker_column_2))::bigint, '0'::bigint))))::bigint, '0'::bigint) DESC
            ->  HashAggregate  (cost=0.00..0.00 rows=0 width=0)
              Group Key: remote_scan.page_id
              ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)
                Task Count: 32
                Tasks Shown: One of 32
                ->  Task
                  Node: host=localhost port=9701 dbname=postgres
                  ->  HashAggregate  (cost=54.70..56.70 rows=200 width=12)
                    Group Key: page_id
                    ->  Seq Scan on page_views_102008 page_views  (cost=0.00..43.47 rows=2247 width=4)
      Task Count: 32
      Tasks Shown: One of 32
      ->  Task
        Node: host=localhost port=9701 dbname=postgres
        ->  HashAggregate  (cost=84.50..86.75 rows=225 width=36)
          Group Key: page_views.page_id, page_views.host_ip
          ->  Hash Join  (cost=17.00..78.88 rows=1124 width=36)
            Hash Cond: (page_views.page_id = intermediate_result.page_id)
            ->  Seq Scan on page_views_102008 page_views  (cost=0.00..43.47 rows=2247 width=36)
            ->  Hash  (cost=14.50..14.50 rows=200 width=4)
              ->  HashAggregate  (cost=12.50..14.50 rows=200 width=4)
                Group Key: intermediate_result.page_id
                ->  Function Scan on read_intermediate_result intermediate_result  (cost=0.00..10.00 rows=1000 width=4)

GroupAggregate  (cost=0.00..0.00 rows=0 width=0)
  Group Key: remote_scan.page_id
  ->  Sort  (cost=0.00..0.00 rows=0 width=0)
    Sort Key: remote_scan.page_id

->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)
  ->  Distributed Subplan 6_1
.

->  Limit  (cost=0.00..0.00 rows=0 width=0)
  ->  Sort  (cost=0.00..0.00 rows=0 width=0)
    Sort Key: COALESCE((pg_catalog.sum((COALESCE((pg_catalog.sum(remote_scan.worker_column_2))::bigint, '0'::bigint))))::bigint, '0'::bigint) DESC
    ->  HashAggregate  (cost=0.00..0.00 rows=0 width=0)
      Group Key: remote_scan.page_id
      ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)
        Task Count: 32
        Tasks Shown: One of 32
        ->  Task
          Node: host=localhost port=9701 dbname=postgres
          ->  HashAggregate  (cost=54.70..56.70 rows=200 width=12)
            Group Key: page_id
            ->  Seq Scan on page_views_102008 page_views  (cost=0.00..43.47 rows=2247 width=4)
.

Task Count: 32
Tasks Shown: One of 32
->  Task
  Node: host=localhost port=9701 dbname=postgres
  ->  HashAggregate  (cost=84.50..86.75 rows=225 width=36)
    Group Key: page_views.page_id, page_views.host_ip
    ->  Hash Join  (cost=17.00..78.88 rows=1124 width=36)
      Hash Cond: (page_views.page_id = intermediate_result.page_id)
.

->  Seq Scan on page_views_102008 page_views  (cost=0.00..43.47 rows=2247 width=36)
->  Hash  (cost=14.50..14.50 rows=200 width=4)
  ->  HashAggregate  (cost=12.50..14.50 rows=200 width=4)
    Group Key: intermediate_result.page_id
    ->  Function Scan on read_intermediate_result intermediate_result  (cost=0.00..10.00 rows=1000 width=4)
.

Когда пользователь отправляет запрос, узел-координатор citus разделяет его на более мелкие фрагменты, где каждый фрагмент может выполняться независимо на рабочем узле. Это позволяет citus распределять каждый запрос по кластеру.

Однако способ разделения запросов на фрагменты (и какие запросы транслируются) зависит от типа запроса. В некоторых сложных ситуациях полезно управлять этим поведением вручную. В citus есть служебные функции для трансляции SQL на рабочие узлы, сегменты или места совмещения.

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

-- Вывод параметра work_mem каждой БД на рабочих узлах
SELECT run_command_on_workers($cmd$ SHOW work_mem; $cmd$);

-- Получить расчётное количество строк для распределённой таблицы, суммируя
-- расчётное количество строк для каждого сегмента.
SELECT sum(result::bigint) AS estimated_count
  FROM run_command_on_shards(
    'my_distributed_table',
    $cmd$
      SELECT reltuples
        FROM pg_class c
        JOIN pg_catalog.pg_namespace n on n.oid=c.relnamespace
       WHERE (n.nspname || '.' || relname)::regclass = '%s'::regclass
         AND n.nspname NOT IN ('citus', 'pg_toast', 'pg_catalog')
    $cmd$
  );

-- Предположим, есть две распределённые таблицы
CREATE TABLE little_vals (key int, val int);
CREATE TABLE big_vals    (key int, val int);
SELECT create_distributed_table('little_vals', 'key');
SELECT create_distributed_table('big_vals',    'key');

-- Необходимо синхронизировать их, чтобы каждый раз при создании
-- little_vals, также создавалось значение big_vals, равное удвоенному значению little_vals
--
-- Сначала создайте триггерную функцию, которая будет
-- принимать в качестве аргумента размещение целевой таблицы
CREATE OR REPLACE FUNCTION embiggen() RETURNS TRIGGER AS $$
  BEGIN
    IF (TG_OP = 'INSERT') THEN
      EXECUTE format(
        'INSERT INTO %s (key, val) SELECT ($1).key, ($1).val*2;',
        TG_ARGV[0]
      ) USING NEW;
    END IF;
    RETURN NULL;
  END;
$$ LANGUAGE plpgsql;

-- Затем свяжите совмещённые таблицы с помощью триггерной функции
-- на каждом размещении
SELECT run_command_on_colocated_placements(
  'little_vals',
  'big_vals',
  $cmd$
    CREATE TRIGGER after_insert AFTER INSERT ON %s
      FOR EACH ROW EXECUTE PROCEDURE embiggen(%L)
  $cmd$
);

Поскольку citus является расширением Postgres Pro, он совместим с её конструкциями. Пользователи могут применять инструменты и функциональность всей экосистемы Postgres Pro для распределённых таблиц, созданных с помощью citus.

В citus реализована 100% поддержка SQL для любых запросов, которые можно выполнить на одном рабочем узле. Запросы такого типа часто встречаются в многоарендных приложениях при доступе к информации об одном арендаторе.

Даже межузловые запросы (используемые для параллельных вычислений) поддерживают большую часть функциональности SQL. Однако некоторая часть функциональности SQL не поддерживается для запросов, объединяющих информацию из нескольких узлов.

SELECT * FROM dist WHERE EXISTS (SELECT 1 FROM local WHERE local.a = dist.a);
/*
ERROR:  direct joins between distributed and local tables are not supported
HINT:  Use CTEs or subqueries to select from local tables and use them in joins
*/

WITH cte AS (SELECT * FROM dist)
SELECT * FROM cte WHERE EXISTS (SELECT 1 FROM local WHERE local.a = cte.a);

-- Такой запрос не работает

  SELECT repo_id, event_type, event_public,
         grouping(event_type, event_public),
         min(created_at)
    FROM github_events
   WHERE repo_id IN (8514, 15435, 19438, 21692)
GROUP BY repo_id, ROLLUP(event_type, event_public);

ERROR:  could not run distributed query with GROUPING
HINT:  Consider using an equality filter on the distributed table's partition column.

-- Размещение данных без агрегирования в локальную таблицу

CREATE TEMP TABLE results AS (
  SELECT repo_id, event_type, event_public, created_at
    FROM github_events
       WHERE repo_id IN (8514, 15435, 19438, 21692)
    );

-- Запуск агрегирования локально

  SELECT repo_id, event_type, event_public,
         grouping(event_type, event_public),
         min(created_at)
    FROM results
GROUP BY repo_id, ROLLUP(event_type, event_public);

 repo_id |    event_type     | event_public | grouping |         min
---------+-------------------+--------------+----------+---------------------
    8514 | PullRequestEvent  | t            |        0 | 2016-12-01 05:32:54
    8514 | IssueCommentEvent | t            |        0 | 2016-12-01 05:32:57
   19438 | IssueCommentEvent | t            |        0 | 2016-12-01 05:48:56
   21692 | WatchEvent        | t            |        0 | 2016-12-01 06:01:23
   15435 | WatchEvent        | t            |        0 | 2016-12-01 05:40:24
   21692 | WatchEvent        |              |        1 | 2016-12-01 06:01:23
   15435 | WatchEvent        |              |        1 | 2016-12-01 05:40:24
    8514 | PullRequestEvent  |              |        1 | 2016-12-01 05:32:54
    8514 | IssueCommentEvent |              |        1 | 2016-12-01 05:32:57
   19438 | IssueCommentEvent |              |        1 | 2016-12-01 05:48:56
   15435 |                   |              |        3 | 2016-12-01 05:40:24
   21692 |                   |              |        3 | 2016-12-01 06:01:23
   19438 |                   |              |        3 | 2016-12-01 05:48:56
    8514 |                   |              |        3 | 2016-12-01 05:32:54

INSERT INTO a.widgets (map_id, widget_name)
VALUES (
    (SELECT mt.map_id FROM a.map_tags mt WHERE mt.map_license = '12345'),
    'Test'
);

INSERT INTO a.widgets (map_id, widget_name)
SELECT mt.map_id, 'Test'
  FROM a.map_tags mt
 WHERE mt.map_license = '12345';

Этот раздел содержит справочную информацию о поддерживаемых в citus пользовательских функциях. Эти функции позволяют использовать в citus расширенные возможности распределения, а не только стандартные SQL-команды.

Существуют различные параметры конфигурации, которые влияют на поведение citus: как стандартные параметры Postgres Pro, так и специальные параметры citus. За подробным описанием параметров конфигурации Postgres Pro обратитесь к главе Настройка сервера.

Оставшаяся часть данной справки посвящена обсуждению особых параметров конфигурации citus. Эти параметры можно задать аналогично параметрам Postgres Pro : с помощью команды SET или изменив файл postgresql.conf.

Например, можно изменить параметр следующим образом:

ALTER DATABASE citus SET citus.multi_task_query_log_level = 'log';

В этом разделе рассматриваются параметры конфигурации для запуска кластера в производственной среде.

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

Архитектура citus на основе логического сегментирования позволяет масштабировать кластер без простоев. В этом разделе описывается добавление новых узлов в кластер citus, чтобы повысить производительность и масштабируемость запросов.

SELECT * from citus_add_node('node-name', 5432);

SELECT citus_rebalance_start();

ALTER SYSTEM SET citus.max_background_task_executors_per_node = 2;
SELECT pg_reload_conf();

SELECT citus_rebalance_start();

-- Допустим у my_table есть уникальный индекс my_table_idx,
-- включающий столбец распределения

ALTER TABLE my_table REPLICA IDENTITY
  USING INDEX my_table_idx;

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

При администрировании кластера citus полезно знать, какие запросы выполняются пользователями, какие узлы задействованы и какой метод выполнения используется citus для каждого запроса. Расширение записывает статистику запросов в представление метаданных citus_stat_statements, названное аналогично представлению pg_stat_statements в Postgres Pro. В представлении pg_stat_statements хранится информация о продолжительности запроса и вводе-выводе, а в citus_stat_statements — о методах выполнения citus и ключах секционирования сегмента (при их наличии).

Чтобы отслеживать статистику запросов в citus, нужно установить расширение pg_stat_statements. В локальном экземпляре Postgres Pro загрузите расширение в postgresql.conf с помощью shared_preload_libraries, затем создайте расширение в SQL:

CREATE EXTENSION pg_stat_statements;

Предположим, есть хеш-распределённая по столбцу id таблица с именем foo.

-- Создайте и заполните распределённую таблицу
CREATE TABLE foo ( id int );
SELECT create_distributed_table('foo', 'id');

INSERT INTO foo SELECT generate_series(1,100);

Выполните ещё два запроса, и citus_stat_statements покажет, каким способом citus выполнил их.

-- Подсчёт всех строк выполняется на всех узлах,
-- результаты суммируются на координаторе
SELECT count(*) FROM foo;

-- Указание строки по столбцу распределения отправляет
-- выполнение на отдельный узел
SELECT * FROM foo WHERE id = 42;

Чтобы узнать, как были выполнены эти запросы, обратитесь к таблице статистики:

SELECT * FROM citus_stat_statements;

Результат:

-[ RECORD 1 ]-+----------------------------------------------
queryid       | -6844578505338488014
userid        | 10
dbid          | 13340
query         | SELECT count(*) FROM foo;
executor      | adaptive
partition_key |
calls         | 1
-[ RECORD 2 ]-+----------------------------------------------
queryid       | 185453597994293667
userid        | 10
dbid          | 13340
query         | INSERT INTO foo SELECT generate_series($1,$2)
executor      | insert-select
partition_key |
calls         | 1
-[ RECORD 3 ]-+----------------------------------------------
queryid       | 1301170733886649828
userid        | 10
dbid          | 13340
query         | SELECT * FROM foo WHERE id = $1
executor      | adaptive
partition_key | 42
calls         | 1

Можно увидеть, что для выполнения запросов citus чаще всего использует адаптивного исполнителя. Он разбивает запрос на несколько запросов для выполнения на соответствующих узлах и объединяет результаты на узле-координаторе. В случае второго запроса (фильтрация по столбцу распределения id = $1) citus определил, что требуются данные только с одного узла. Наконец, можно увидеть, что оператор INSERT INTO foo SELECT… выполняется с помощью исполнителя insert-select, который обеспечивает гибкость при выполнении запросов такого типа.

SELECT sum(calls),
       partition_key IS NOT NULL AS single_tenant
FROM citus_stat_statements
GROUP BY 2;

.
 sum | single_tenant
-----+---------------
   2 | f
   1 | t

Расширение citus обеспечивает Postgres Pro функциональностью распределения с помощью программных интерфейсов и обработчика. Эта функциональность включает, помимо прочего, поддержку широкого спектра типов данных (включая полуструктурированные типы данных, такие как jsonb и hstore), операторы и функции, полнотекстовый поиск и другие расширения, такие как PostGIS и HyperLogLog. Кроме того, правильное использование программных интерфейсов расширения обеспечивает совместимость со стандартными инструментами Postgres Pro, такими как pgAdmin и pg_upgrade.

Поскольку расширение citus можно установить на любой экземпляр Postgres Pro, вместе с ним также могут напрямую использоваться другие расширения, такие как hstore, hll или PostGIS. Однако следует помнить об одной особенности. При включении других расширений в shared_preload_libraries citus всегда должен указываться первым.

Следующие расширения будут особенно полезны при работе с citus:

Каждый сервер Postgres Pro может содержать несколько баз данных. Однако новые базы данных не наследуют расширения других; все нужные расширения необходимо добавлять заново. Чтобы запустить citus в новой базе данных, нужно создать базу данных на узле-координаторе и рабочих узлах, создать расширение citus в этой базе данных и зарегистрировать рабочие узлы в базе данных узла-координатора.

Подключитесь к каждому рабочему узлу и выполните:

-- На каждом рабочем узле

CREATE DATABASE newbie;
\c newbie
CREATE EXTENSION citus;

Затем выполните на узле-координаторе:

CREATE DATABASE newbie;
\c newbie
CREATE EXTENSION citus;

SELECT * from citus_add_node('node-name', 5432);
SELECT * from citus_add_node('node-name2', 5432);
-- ... для всех узлов

Теперь новая база данных будет работать как ещё один кластер citus.

Обычный способ определения размеров таблиц в Postgres Pro, pg_total_relation_size, существенно занижает размер распределённых таблиц. В кластере citus эта функция только показывает размер таблиц на узле-координаторе. На самом деле данные в распределённых таблицах находятся на рабочих узлах (сегментах), а не на узле-координаторе. Реальный размер распределённой таблицы складывается из размеров сегментов. В citus доступны вспомогательные функции для получения этой информации.

Эти функции аналогичны трём стандартным функциям получения размера объектов БД в Postgres Pro, с небольшим дополнением: если они не могут подключиться к узлу, выдаётся ошибка.

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

SELECT logicalrelid AS name,
       pg_size_pretty(citus_table_size(logicalrelid)) AS size
  FROM pg_dist_partition;

Результат:

┌───────────────┬───────┐
│     name      │ size  │
├───────────────┼───────┤
│ github_users  │ 39 MB │
│ github_events │ 37 MB │
└───────────────┴───────┘

В Postgres Pro (и других базах данных MVCC) команды UPDATE или DELETE не приводят к немедленному удалению старой версии строки. Накопление устаревших строк называется раздуванием, и от них необходимо избавляться, чтобы избежать снижения производительности запросов и неконтролируемого заполнения дискового пространства. Postgres Pro запускает процесс, называемый демоном автоочистки, который периодически очищает (удаляет) устаревшие строки.

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

В связи с вышеизложенным операции автоочистки в кластере citus должны быть эффективны в большинстве случаев. Однако для таблиц с определёнными нагрузками или для компаний с установленными «безопасными» часами очистки может быть целесообразнее очищать таблицу вручную, а не оставлять всю работу автоочистке.

Чтобы очистить таблицу, выполните на узле-координаторе:

VACUUM my_distributed_table;

При очистке распределённой таблицы команда VACUUM будет отправлена во все места размещения этой таблицы (одно подключение на каждое место размещения). Это делается параллельно. Поддерживаются все параметры (в том числе список таблица_и_столбцы), за исключением VERBOSE. Команда VACUUM также запускается на узле-координаторе до уведомления каких-либо рабочих узлов. Обратите внимание, что неквалифицированные команды очистки (т. е. команды без указанной таблицы) не транслируются на рабочие узлы.

Команда Postgres Pro ANALYZE собирает статистику о содержимом таблиц в базе данных. Впоследствии планировщик запросов использует эту статистику, чтобы определить наиболее эффективные планы выполнения запросов.

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

Чтобы проанализировать таблицу, выполните на узле-координаторе:

ANALYZE my_distributed_table;

В citus команда ANALYZE транслируется на все размещения рабочих узлов.

В citus реализовано хранилище столбцовых таблиц с возможностью только добавления для аналитики и хранения данных. Когда столбцы (а не строки) хранятся на диске последовательно, данные становятся более сжимаемыми, и запросы могут быстрее обращаться к подмножеству столбцов.

CREATE TABLE contestant (
    handle TEXT,
    birthdate DATE,
    rating INT,
    percentile FLOAT,
    country CHAR(3),
    achievements TEXT[]
) USING columnar;

-- Преобразование в строковое хранение (метод доступа heap)
SELECT alter_table_set_access_method('contestant', 'heap');

-- Преобразование в столбцовое хранение (индексы будут удалены)
SELECT alter_table_set_access_method('contestant', 'columnar');

-- Все эти значения вставляются в один столбцовый массив

INSERT INTO contestant VALUES
  ('a','1990-01-10',2090,97.1,'XA','{a}'),
  ('b','1990-11-01',2203,98.1,'XA','{a,b}'),
  ('c','1988-11-01',2907,99.4,'XB','{w,y}'),
  ('d','1985-05-05',2314,98.3,'XB','{}'),
  ('e','1995-05-05',2236,98.2,'XC','{a}');

VACUUM VERBOSE contestant;

INFO:  statistics for "contestant":
storage id: 10000000000
total file size: 24576, total data size: 248
compression rate: 1.31x
total row count: 5, stripe count: 1, average rows per stripe: 5
chunk count: 6, containing data for dropped columns: 0, zstd compressed: 6

-- Сначала создайте большую таблицу со строковым хранением.
CREATE TABLE perf_row(
  c00 int8, c01 int8, c02 int8, c03 int8, c04 int8, c05 int8, c06 int8, c07 int8, c08 int8, c09 int8,
  c10 int8, c11 int8, c12 int8, c13 int8, c14 int8, c15 int8, c16 int8, c17 int8, c18 int8, c19 int8,
  c20 int8, c21 int8, c22 int8, c23 int8, c24 int8, c25 int8, c26 int8, c27 int8, c28 int8, c29 int8,
  c30 int8, c31 int8, c32 int8, c33 int8, c34 int8, c35 int8, c36 int8, c37 int8, c38 int8, c39 int8,
  c40 int8, c41 int8, c42 int8, c43 int8, c44 int8, c45 int8, c46 int8, c47 int8, c48 int8, c49 int8,
  c50 int8, c51 int8, c52 int8, c53 int8, c54 int8, c55 int8, c56 int8, c57 int8, c58 int8, c59 int8,
  c60 int8, c61 int8, c62 int8, c63 int8, c64 int8, c65 int8, c66 int8, c67 int8, c68 int8, c69 int8,
  c70 int8, c71 int8, c72 int8, c73 int8, c74 int8, c75 int8, c76 int8, c77 int8, c78 int8, c79 int8,
  c80 int8, c81 int8, c82 int8, c83 int8, c84 int8, c85 int8, c86 int8, c87 int8, c88 int8, c89 int8,
  c90 int8, c91 int8, c92 int8, c93 int8, c94 int8, c95 int8, c96 int8, c97 int8, c98 int8, c99 int8
);

-- Затем создайте таблицу с идентичными столбцами, но столбцовым хранением.
CREATE TABLE perf_columnar(LIKE perf_row) USING COLUMNAR;

INSERT INTO perf_row
  SELECT
    g % 00500, g % 01000, g % 01500, g % 02000, g % 02500, g % 03000, g % 03500, g % 04000, g % 04500, g % 05000,
    g % 05500, g % 06000, g % 06500, g % 07000, g % 07500, g % 08000, g % 08500, g % 09000, g % 09500, g % 10000,
    g % 10500, g % 11000, g % 11500, g % 12000, g % 12500, g % 13000, g % 13500, g % 14000, g % 14500, g % 15000,
    g % 15500, g % 16000, g % 16500, g % 17000, g % 17500, g % 18000, g % 18500, g % 19000, g % 19500, g % 20000,
    g % 20500, g % 21000, g % 21500, g % 22000, g % 22500, g % 23000, g % 23500, g % 24000, g % 24500, g % 25000,
    g % 25500, g % 26000, g % 26500, g % 27000, g % 27500, g % 28000, g % 28500, g % 29000, g % 29500, g % 30000,
    g % 30500, g % 31000, g % 31500, g % 32000, g % 32500, g % 33000, g % 33500, g % 34000, g % 34500, g % 35000,
    g % 35500, g % 36000, g % 36500, g % 37000, g % 37500, g % 38000, g % 38500, g % 39000, g % 39500, g % 40000,
    g % 40500, g % 41000, g % 41500, g % 42000, g % 42500, g % 43000, g % 43500, g % 44000, g % 44500, g % 45000,
    g % 45500, g % 46000, g % 46500, g % 47000, g % 47500, g % 48000, g % 48500, g % 49000, g % 49500, g % 50000
  FROM generate_series(1,50000000) g;

INSERT INTO perf_columnar
  SELECT
    g % 00500, g % 01000, g % 01500, g % 02000, g % 02500, g % 03000, g % 03500, g % 04000, g % 04500, g % 05000,
    g % 05500, g % 06000, g % 06500, g % 07000, g % 07500, g % 08000, g % 08500, g % 09000, g % 09500, g % 10000,
    g % 10500, g % 11000, g % 11500, g % 12000, g % 12500, g % 13000, g % 13500, g % 14000, g % 14500, g % 15000,
    g % 15500, g % 16000, g % 16500, g % 17000, g % 17500, g % 18000, g % 18500, g % 19000, g % 19500, g % 20000,
    g % 20500, g % 21000, g % 21500, g % 22000, g % 22500, g % 23000, g % 23500, g % 24000, g % 24500, g % 25000,
    g % 25500, g % 26000, g % 26500, g % 27000, g % 27500, g % 28000, g % 28500, g % 29000, g % 29500, g % 30000,
    g % 30500, g % 31000, g % 31500, g % 32000, g % 32500, g % 33000, g % 33500, g % 34000, g % 34500, g % 35000,
    g % 35500, g % 36000, g % 36500, g % 37000, g % 37500, g % 38000, g % 38500, g % 39000, g % 39500, g % 40000,
    g % 40500, g % 41000, g % 41500, g % 42000, g % 42500, g % 43000, g % 43500, g % 44000, g % 44500, g % 45000,
    g % 45500, g % 46000, g % 46500, g % 47000, g % 47500, g % 48000, g % 48500, g % 49000, g % 49500, g % 50000
  FROM generate_series(1,50000000) g;

VACUUM (FREEZE, ANALYZE) perf_row;
VACUUM (FREEZE, ANALYZE) perf_columnar;

SELECT pg_total_relation_size('perf_row')::numeric/
       pg_total_relation_size('perf_columnar') AS compression_ratio;

.
 compression_ratio
--------------------
 8.0196135873627944
(1 row)

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

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

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

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

Координатор citus разбивает входящий запрос на фрагменты и отправляет их рабочим узлам для параллельной обработки. Рабочие узлы — это просто расширенные серверы Postgres Pro, применяющие стандартную логику планирования и выполнения Postgres Pro для этих запросов. Итак, первый шаг в настройке citus — настройка параметров конфигурации Postgres Pro на рабочих узлах для обеспечения высокой производительности.

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

Чтобы начать процесс настройки, создайте кластер citus и загрузите в него данные. На узле-координаторе запустите команду EXPLAIN для репрезентативных запросов, чтобы проверить производительность. В citus команда EXPLAIN также предоставляет информацию о выполнении распределённых запросов. Вывод EXPLAIN показывает, как каждый рабочий узел обрабатывает запрос, а также как узел-координатор объединяет их результаты.

Ниже приведён пример объяснения плана для конкретного примера запроса. Используется флаг VERBOSE, чтобы увидеть фактические запросы, которые были отправлены на рабочие узлы.

EXPLAIN VERBOSE
 SELECT date_trunc('minute', created_at) AS minute,
        sum((payload->>'distinct_size')::int) AS num_commits
   FROM github_events
  WHERE event_type = 'PushEvent'
  GROUP BY minute
  ORDER BY minute;

Sort  (cost=0.00..0.00 rows=0 width=0)
  Sort Key: remote_scan.minute
  ->  HashAggregate  (cost=0.00..0.00 rows=0 width=0)
    Group Key: remote_scan.minute
    ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)
      Task Count: 32
      Tasks Shown: One of 32
      ->  Task
        Query: SELECT date_trunc('minute'::text, created_at) AS minute, sum(((payload OPERATOR(pg_catalog.->>) 'distinct_size'::text))::integer) AS num_commits FROM github_events_102042 github_events WHERE (event_type OPERATOR(pg_catalog.=) 'PushEvent'::text) GROUP BY (date_trunc('minute'::text, created_at))
        Node: host=localhost port=5433 dbname=postgres
        ->  HashAggregate  (cost=93.42..98.36 rows=395 width=16)
          Group Key: date_trunc('minute'::text, created_at)
          ->  Seq Scan on github_events_102042 github_events  (cost=0.00..88.20 rows=418 width=503)
            Filter: (event_type = 'PushEvent'::text)
(13 rows)

Результат говорит о следующем. Есть 32 сегмента, и планировщик выбрал адаптивного исполнителя citus для выполнения этого запроса:

->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)
  Task Count: 32

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

Tasks Shown: One of 32
->  Task
  Query: SELECT date_trunc('minute'::text, created_at) AS minute, sum(((payload OPERATOR(pg_catalog.->>) 'distinct_size'::text))::integer) AS num_commits FROM github_events_102042 github_events WHERE (event_type OPERATOR(pg_catalog.=) 'PushEvent'::text) GROUP BY (date_trunc('minute'::text, created_at))
  Node: host=localhost port=5433 dbname=postgres