pgstream - сбор данных об изменениях в реальном времени для PostgreSQL

Как мы уже говорили, Postgres подходит для всего, для всех стеков. Теперь к этому списку добавился и Change Data Capture, благодаря pgstream.

Когда мы говорим «стеки», мы имеем в виду стек данных, стек ML, стек временных рядов, управление пространственными данными, потоковую обработку в реальном времени, ETL и так далее. PostgreSQL больше не является просто СУБД, а превратился в швейцарский армейский нож, который может делать практически все.

Pgstream - это отдельная утилита, а не расширение для Postgres, и создана она ребятами из Xata, бессерверной платформы данных, построенной на базе PostgreSQL и предоставляющей, помимо прочего, полнотекстовый и векторный поисковый механизм, прикрепление файлов на уровне записей и агрегацию на уровне таблиц.

Официальное определение pgstream гласит, что он облегчает репликацию Postgres с изменениями DDL. Что это значит?

• Репликация изменений DDL
  Беспрепятственное отслеживание и репликация изменений схемы, включая DDL, для улучшенной репликации Postgres в любой пункт назначения вывода.
• Вывод данных Elasticsearch и OpenSearch
  Автоматическая репликация данных и изменений схемы Postgres в Elasticsearch и Opensearch.
• Модульная конфигурация развертывания
  pgstream - это готовое решение для PostgreSQL CDC, которое легко расширяется с помощью модулей трансформации или очереди сообщений перед выводом в нужное вам место.
• Webhook ouput
  Запускайте конечную точку webhook по вашему выбору мгновенно при каждом изменении схемы или данных Postgres.

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

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

Что касается функции полнотекстового поиска, которая также встроена в Xata, то она использует Elasticsearch поверх Postgres. Чтобы синхронизировать эти два хранилища данных, он перехватывает и идентифицирует изменения данных и схемы в Postgres и с минимальной задержкой передает эти изменения в Elasticsearch.

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

Срабатывание webhook происходит после фиксации DDL, поскольку оно основано на триггере, который заполняет таблицу pgstream.schema_log схемой в формате JSON, а это должно происходить только после фиксации операции. Эта запись в таблице pgstream.schema_log затем реплицируется вниз и обрабатывается webhook notifier для отправки уведомлений на подписанные webhook URL.

Итак, когда выполняется следующий DDL:

INSERT INTO employees(name, role) VALUES('Dwight Schrute', 'Assistant Regional Manager');

Вот что создается и передается в Webhook.

Так просто. По сравнению с тем, через что мне пришлось пройти, когда я вручную реализовывал функциональность CDC/Webhook, которая включала сложный конвейер триггеров и событий базы данных в Ingres dbms для преобразования и манипулирования данными, собранными SQL-запросами, в пакеты HL7 перед их автоматической отправкой в веб-сервис в конце конвейера.

Достаточно сказано, давайте посмотрим, как можно начать работу с pgstream. Для этого вам понадобится экземпляр Postgres с включенной логической репликацией и загруженным плагином вывода wal2json. Есть несколько способов сделать это, но наиболее простым является использование файла docker-compose из репозитория pgstream:

git clone https://github.com/xataio/pgstream.git

Затем запустите экземпляр Postgres:

docker compose -f build/docker/docker-compose.yml up db

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

psql «postgres://postgres:postgres@localhost:5432/postgres»;

Теперь установите pgstream. Поскольку у нас уже есть клонированное репо, мы собираем его из исходников:

go build

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

./pgstream init --pgurl «postgres://postgres:postgres@localhost/postgres?sslmode=disable»;

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

Помимо формата утилиты, pgstream также предлагается в виде библиотеки, вызываемой из кода на Go. Официальной документации по ней пока нет, но в качестве примера того, как она используется внутри pgstream CLI в stream pkg, можно посмотреть этот официальный пример.

Итак, вот и все. Полнофункциональный CDC в реальном времени, простой и бесплатный, с помощью pgstream!