Skip to content

Установка

Требования

Компонент Версия
Greenplum 6.27.1-p2 (PostgreSQL 9.4.26)
librdkafka 1.9 и новее, на координаторе и всех сегментах
Apache Kafka 2.x и новее
Платформа rhel7-gcc11, rhel8, ubuntu22.04

Коннектор ставится поверх работающего кластера Greenplum. Отдельного сервиса он не поднимает и PXF ему не нужен - расширение линкуется с librdkafka и ходит в брокеры само.

Шаг 1. librdkafka на всех хостах

Библиотека должна лежать на координаторе и на каждом сегмент-хосте, иначе сегменты не смогут загрузить kafka_fdw.so.

# Ubuntu 22.04
apt-get install -y librdkafka1

# CentOS 7 (репозиторий Confluent)
rpm --import https://packages.confluent.io/rpm/7.5/archive.key
printf '[Confluent]\nname=Confluent\nbaseurl=https://packages.confluent.io/rpm/7.5\ngpgcheck=1\ngpgkey=https://packages.confluent.io/rpm/7.5/archive.key\nenabled=1\n' > /etc/yum.repos.d/confluent.repo
yum install -y librdkafka

Для CentOS 8 готовых пакетов Confluent нет - библиотека собирается из исходников:

dnf install -y gcc make openssl-devel cyrus-sasl-devel zlib-devel
curl -sL https://github.com/confluentinc/librdkafka/archive/refs/tags/v2.3.0.tar.gz | tar -xz -C /tmp
cd /tmp/librdkafka-2.3.0 && ./configure --prefix=/usr && make -j"$(nproc)" && make install && ldconfig

Шаг 2. Получить дистрибутив

Пакеты лежат в Nexus, отдельно под каждую платформу:

https://repo.datasapience.ru/repository/gp_raw/components/kafka_fdw/kafka_fdw-1.5.0-<платформа>-x86_64.tar.gz

Анонимный запрос вернёт 401, скачивание требует базовой авторизации:

curl -u '<логин>:<пароль>' -O \
    "https://repo.datasapience.ru/repository/gp_raw/components/kafka_fdw/kafka_fdw-1.5.0-ubuntu22.04-x86_64.tar.gz"
tar -xzf kafka_fdw-1.5.0-ubuntu22.04-x86_64.tar.gz

Логин и пароль на чтение репозитория узнайте у того, кто выдаёт доступ к Nexus в вашей организации.

Шаг 3. Установка через Ansible

В архиве лежит готовая роль kafka_fdw_install. Управляющему хосту нужен ansible-core и ssh-доступ ко всем хостам Greenplum.

cd kafka_fdw-1.5.0-ubuntu22.04-x86_64
ansible-galaxy collection install -r requirements.yml
cp inventory.example inventory

Опишите в inventory свой кластер:

[master]
gp-mstr ansible_host=192.168.1.11 ansible_ssh_user=gpadmin

[segments]
gp-seg1 ansible_host=192.168.1.21 ansible_ssh_user=gpadmin
gp-seg2 ansible_host=192.168.1.22 ansible_ssh_user=gpadmin

[masters:children]
master

[greenplum_cluster:children]
masters
segments

[all:vars]
gphome=/usr/local/greenplum-db
cluster_admin_user=gpadmin
database_name=gpdb

Запуск:

# Полная установка: файлы на все хосты + CREATE EXTENSION
ansible-playbook -i inventory playbook.yml

# Только файлы, без CREATE EXTENSION
ansible-playbook -i inventory playbook.yml --tags install

# Только CREATE EXTENSION, файлы уже разложены
ansible-playbook -i inventory playbook.yml --tags configure

Плейбук раскладывает kafka_fdw.so и файлы расширения в $GPHOME на координаторе и всех сегмент-хостах, а затем выполняет CREATE EXTENSION IF NOT EXISTS kafka_fdw в базе database_name. Повторный запуск безопасен.

Останавливать или перезапускать Greenplum установка не требует. Новую библиотеку подхватит любая свежая сессия; уже открытые соединения продолжат работать со старой версией до переподключения.

Шаг 4. Создание расширения в базе

Если плейбук отработал целиком, шаг уже выполнен. Вручную:

CREATE EXTENSION kafka_fdw;

Расширение регистрирует foreign data wrapper kafka_fdw и схему kafka с функциями мониторинга topic_info(), consumer_lag(), offsets_info(), version() и функциями управления офсетами set_offset(), reset_offsets(). Таблицы офсетов оно не создаёт - начиная с версии 1.3 офсеты живут в самой Kafka.

Проверка после установки:

SELECT extversion FROM pg_extension WHERE extname = 'kafka_fdw';   -- 1.5
SELECT kafka.version();                                            -- 1.5.0

Установка вручную

Если Ansible недоступен, файлы раскладываются руками. Сборка на координаторе:

make PG_CONFIG=$GPHOME/bin/pg_config
sudo make install PG_CONFIG=$GPHOME/bin/pg_config

Затем копирование на каждый сегмент-хост:

for host in gp-seg1 gp-seg2; do
  scp $GPHOME/lib/postgresql/kafka_fdw.so                $host:$GPHOME/lib/postgresql/
  scp $GPHOME/share/postgresql/extension/kafka_fdw*      $host:$GPHOME/share/postgresql/extension/
done

После этого CREATE EXTENSION kafka_fdw в базе, как в шаге 4.

Библиотека должна быть одинаковой на всех хостах. Устаревшая .so хотя бы на одном сегменте даст ошибки старых версий - причём воспроизводиться они будут через раз, в зависимости от того, какому сегменту достанется партиция. Сверить проще всего из SQL, kafka.version() отвечает с того хоста, где выполняется:

SELECT 'master' AS host, kafka.version()
UNION ALL
SELECT 'seg' || gp_segment_id, kafka.version() FROM gp_dist_random('gp_id')
ORDER BY 1;

Все строки должны показать одну версию. На версиях до 1.5 остаются контрольные суммы:

for h in gp-mstr gp-seg1 gp-seg2; do ssh $h md5sum $GPHOME/lib/postgresql/kafka_fdw.so; done