nextdoor/pg-bifrost

pg-bifrost

镜像概述和主要用途

pg-bifrost 是基于 PostgreSQL 逻辑解码 的工具，用于捕获数据库事件流（包括创建、插入、更新、删除等操作），并将其写入 Amazon Kinesis Data Streams 或 RabbitMQ。其模块化设计支持扩展至 S3、DynamoDB 等其他存储或非 AWS 目标，适用于实时数据同步、数据流处理和变更捕获场景。

核心功能和特性

• 逻辑解码支持：通过 PostgreSQL 逻辑解码机制捕获数据库变更事件
• 多目标集成：原生支持 Kinesis Data Streams 和 RabbitMQ，可扩展至其他存储目标
• 灵活配置：支持通过命令行参数、环境变量或配置文件进行配置
• 数据过滤：提供表级别的白名单和黑名单过滤功能
• 分区策略：多种分区方法确保数据有序性和吞吐量优化
• 批量处理：可配置的批量刷新策略，平衡延迟和吞吐量

安装

可通过以下方式获取 pg-bifrost：

• 预编译二进制：从 GitHub https://github.com/Nextdoor/pg-bifrost/releases 下载 Linux AMD64 版本
• Docker 镜像：Docker Hub 地址 https://hub.docker.com/r/nextdoor/pg-bifrost%EF%BC%88%E7%94%B1 CircleCI 自动构建部署）

使用示例

启动命令

bash
[Slava pg-bifrost.git]$ export AWS_SECRET_ACCESS_KEY=secretaccesskey
[Slava pg-bifrost.git]$ export AWS_ACCESS_KEY_ID=accesskeyid
[Slava pg-bifrost.git]$ export AWS_REGION=us-east-1
[Slava pg-bifrost.git]$ pg-bifrost --dbname mydb --host 127.0.0.1 replicate --create-slot kinesis --kinesis-stream dbstream

数据库操作示例（输入）

sql
CREATE TABLE customers (id serial primary key, first_name text, last_name text);

INSERT INTO customers (first_name, last_name) VALUES ('Hello', 'World');
INSERT INTO customers (first_name, last_name) VALUES ('Goodbye', 'World');
UPDATE customers SET last_name = 'Friends' where first_name = 'Hello';
DELETE FROM customers WHERE first_name = 'Goodbye';

输出事件流

json
{"time":"1970-01-01T00:00:01Z","lsn":"0/1510A58","table":"public.customers","operation":"INSERT","columns":{"first_name":{"new":{"q":"true","t":"text","v":"Hello"}},"id":{"new":{"q":"false","t":"integer","v":"1"}},"last_name":{"new":{"q":"true","t":"text","v":"World"}}}}
{"time":"1970-01-01T00:00:01Z","lsn":"0/1510B60","table":"public.customers","operation":"INSERT","columns":{"first_name":{"new":{"q":"true","t":"text","v":"Goodbye"}},"id":{"new":{"q":"false","t":"integer","v":"2"}},"last_name":{"new":{"q":"true","t":"text","v":"World"}}}}
{"time":"1970-01-01T00:00:01Z","lsn":"0/1510C20","table":"public.customers","operation":"UPDATE","columns":{"first_name":{"new":{"q":"true","t":"text","v":"Hello"}},"id":{"new":{"q":"false","t":"integer","v":"1"}},"last_name":{"new":{"q":"true","t":"text","v":"Friends"}}}}
{"time":"1970-01-01T00:00:01Z","lsn":"0/1510CA8","table":"public.customers","operation":"DELETE","columns":{"id":{"old":{"q":"false","t":"integer","v":"2"}}}}

使用方法

命令概览

USAGE:
   pg-bifrost [global options] command [command options] [arguments...]

COMMANDS:
     create, c     创建复制槽
     drop, d       删除复制槽
     replicate, r  启动逻辑复制
     help, h       显示命令列表或单个命令帮助

GLOBAL OPTIONS:
   --config value    bifrost YAML 配置文件路径（默认："config.yaml"）
   --slot value      PostgreSQL 复制槽名称（默认："pg_bifrost"） [$REPLICATION_SLOT]
   --user value      PostgreSQL 复制用户（默认："replication"） [$PGUSER]
   --password value  PostgreSQL 复制用户密码 [$PGPASSWORD]
   --host value      PostgreSQL 连接主机（默认："127.0.0.1"） [$PGHOST]
   --port value      PostgreSQL 连接端口（默认："5432"） [$PGPORT]
   --dbname value    PostgreSQL 数据库名称（默认："postgres"） [$PGDATABASE]
   --help, -h        显示帮助
   --version, -v     打印版本信息

replicate 命令详情

NAME:
   pg-bifrost replicate - 启动逻辑复制

USAGE:
   pg-bifrost replicate command [command options] [arguments...]

COMMANDS:
     stdout    复制到标准输出
     kinesis   复制到 Kinesis
     rabbitmq  复制到 RabbitMQ

OPTIONS:
   --create-slot, -s                  复制前若复制槽不存在则创建 [$CREATE_SLOT]
   --workers value                    传输工作线程数（默认：1） [$WORKERS]
   --client-buffer-size value         从 PostgreSQL 缓冲的消息数（默认：10000） [$CLIENT_BUFFER_SIZE]
   --batch-flush-update-age value     批处理等待新消息的毫秒数（收到新消息时重置计时器，每秒评估一次）（默认：500） [$BATCH_FLUSH_UPDATE_AGE]
   --batch-flush-max-age value        批处理最大等待毫秒数（覆盖 batch-flush-update-age，每秒评估一次）（默认：1000） [$BATCH_FLUSH_MAX_AGE]
   --batch-queue-depth value          每个工作线程可排队的批处理数（默认：2） [$BATCH_QUEUE_DEPTH]
   --batcher-memory-soft-limit value  批处理消息的最大内存使用量（仅在 batch-flush-timeout 时评估，单位：字节）（默认：104857600） [$BATCHER_MEMORY_SOFT_LIMIT]
   --batcher-routing-method value     批处理路由至工作线程的方式："round-robin"（默认，轮询）或 "partition"（分区，需严格有序时使用）（默认："round-robin"） [$BATCHER_ROUTING_METHOD]
   --partition-method value           消息分区方式："none"（默认，无分区）、"tablename"（按表名）、"transaction"（按事务，单事务一批）、"transaction-bucket"（事务哈希分区）（默认："none"） [$PARTITION_METHOD]
   --batcher-partition-count value    partition-method="transaction-bucket" 时的分区桶数量（默认：1） [$PARTITION_COUNT]
   --whitelist value                  包含的表白名单（其他表将被排除） [$WHITELIST]
   --blacklist value                  排除的表黑名单（其他表将被包含） [$BLACKLIST]
   --help, -h                         显示帮助

配置说明

pg-bifrost 支持通过命令行参数、环境变量或配置文件进行配置，详细选项见 --help。

内存与吞吐量配置

数据过滤

支持表级别的包含/排除过滤：

• 白名单（--whitelist）：仅包含指定表，其他表排除
• 黑名单（--blacklist）：排除指定表，其他表包含
• 注意：白名单和黑名单互斥，不可同时使用

分区策略

分区配置决定消息如何分组和路由，确保数据有序性和吞吐量平衡：

配置选项

策略速查表

开发相关

问题反馈

非安全问题请在 GitHub Issues 提交。

贡献指南

通过以下方式帮助改进项目：

• 报告错误
• 完善文档
• 添加新功能
• 提高测试覆盖率

集成测试

集成测试需 docker 和 docker-compose，步骤如下：

bash
# 拉取 bats 子模块
git submodule sync
git submodule update --init

# 构建 Docker 镜像
make docker_build

# 运行集成测试
make itests

致谢

本工具受 https://github.com/nickelser/pg_kinesis 启发。

许可证

pg-bifrost 基于 Apache License (Version 2.0) 开源，详见 LICENSE。