PG不合群, PG没朋友! PG的臭毛病要改改了

PG不合群, PG没朋友! PG的臭毛病要改改了

说起PG的不合群, 最大的问题就是几乎所有的功能PG都可以通过插件来实现, 比如应用常用的搜索引擎需求、非结构化数据存储需求、向量类型和搜索需求、时序数据支持等. 通过tsvector, 内置JSON, pgvector, timescaledb等插件都能做.

PG是全能多模态数据库选手, 所以和其他产品的交集真的不多, 自然朋友圈就较小了, 不像MySQL的盆友圈那么大, 除了CRUD, 什么都要依赖其他产品.

不过你要对PG这个不合群做出改观了, 今天介绍一个将PG数据实时同步到 Elasticsearch / OpenSearch 搜索引擎的工具: pgsync

下面来详细了解一下 pgsync

https://github.com/toluaina/pgsync

PGSync 架构概览

PGSync 是一个中间件 (middleware)，用于将 PostgreSQL 中的数据实时 (real-time) 同步到 Elasticsearch/OpenSearch 搜索集群中。它的核心目的是让 PostgreSQL 保持为事实来源 (source of truth)（用于事务 (transactions) 和持久化 (persistence)），而 Elasticsearch/OpenSearch 则作为只读的搜索层 (read-only search layer) 和非规范化文档 (denormalized documents) 的存储库。

核心组件和数据流

PGSync 的工作流程依赖于 PostgreSQL 的两个关键功能和几个外部组件 :

工作原理步骤

1. 模式定义：用户创建 schema.json，定义 PostgreSQL 表如何映射和转换 (transform) 成 Elasticsearch/OpenSearch 文档结构。
2. 引导 (Bootstrap): PGSync 在 PostgreSQL 中创建所需的复制槽 (replication slots) 和触发器 (triggers) 以启用变更捕获。
3. 变更捕获 (Change Capture): PGSync Daemon 持续监听 逻辑解码流 (logical decoding stream) 以获取数据变化。
4. 转换 (Transformation): 当接收到变更事件时，Query Builder 执行复杂的 SQL 查询，将涉及多个表和嵌套关系 (nested relationships) 的数据非规范化 (denormalize) 为单个 JSON 文档。
5. 索引 (Indexing): PGSync 将生成的 JSON 文档 批量 (Bulk Index) 写入 Elasticsearch/OpenSearch，实现近实时同步 (near real-time sync)。

关键特性

PostgreSQL/MySQL/MariaDB 到 Elasticsearch/OpenSearch 的同步

PGSync 是一个中间件 (middleware)，用于将数据从 Postgres 或 MySQL 或 MariaDB 轻松同步到 Elasticsearch/OpenSearch。

它允许您将 Postgres 或 MySQL 或 MariaDB 保持为您的事实来源 (source of truth)，并在 Elasticsearch/OpenSearch 中公开结构化的非规范化文档 (structured denormalized documents)。

对嵌套实体 (nested entities) 的更改会被传播到 Elasticsearch/OpenSearch。

PGSync 的高级查询构建器会根据您的模式 (schema) 即时 (on the fly) 生成优化后的 SQL 查询 (optimized SQL queries)。 PGSync 的咨询模型 (advisory model) 允许您在保持关系完整性 (relational integrity) 的同时，快速移动和转换大量数据。

只需在 JSON 中描述您的文档结构 (document structure) 或模式 (schema)，PGSync 就会持续捕获 (continuously capture) 您数据中的更改，并将其加载到 Elasticsearch/OpenSearch 中，无需编写任何代码。 PGSync 将您的关系数据 (relational data) 转换为结构化的文档格式 (structured document format)。

它允许您直接利用 Postgres 或 MySQL 或 MariaDB 的表现力 (expressive power) 和 Elasticsearch/OpenSearch 的可扩展性 (scalability)。 您不必编写复杂的查询和转换管道 (transformation pipelines)。 PGSync 轻量级、灵活且快速。

Elasticsearch/OpenSearch 更适合作为辅助的非规范化搜索引擎 (secondary denormalised search engine)，以配合更传统的规范化数据存储 (normalized datastore)。 此外，您不应该将主数据 (primary data) 存储在 Elasticsearch/OpenSearch 中。

那么，您最初是如何将数据导入 Elasticsearch/OpenSearch 中的呢？ 像 Logstash 和 Kafka 这样的工具可以协助完成这项任务，但它们仍然需要一些工程 (engineering) 和开发 (development) 工作。

抽取、转换、加载 (Extract Transform Load, ETL) 和变更数据捕获 (Change data capture, CDC) 工具可能很复杂，并且需要昂贵的工程投入。

PGSync 的其他优势包括：

• 实时分析 (Real-time analytics)
• 可靠的主数据存储/事实来源 (Reliable primary datastore/source of truth)
• 按需扩展 (Scale on-demand)（多个消费者 (consumers)）
• 轻松连接多个嵌套表 (Easily join multiple nested tables)

从高层次上讲，您的数据位于 PostgreSQL/MySQL/MariaDB 数据库中，并且您希望将其镜像 (mirror) 到 Elasticsearch/OpenSearch 中。 这意味着您数据的每一个变化（Insert、Update、Delete 和 Truncate语句 (statements)）都需要复制 (replicated) 到 Elasticsearch/OpenSearch。 起初，这看起来很容易，但随后就不容易了。只需添加一些代码，在更新数据库后将数据复制到 Elasticsearch/OpenSearch（即所谓的双写 (dual writes)）。 编写跨越多个表 (spanning multiple tables) 并涉及多个关系 (multiple relationships) 的 SQL 查询很难。 检测嵌套文档 (Detecting changes within a nested document) 中的更改也可能相当困难。 当然，如果您的数据从未更改，那么您只需在时间点上拍摄快照 (take a snapshot in time) 并将其作为一次性操作加载到 Elasticsearch/OpenSearch 中即可。

如果出现以下情况，PGSync 适合您：

• Postgres 或 MySQL 或 MariaDB 是您的读/写事实来源 (read/write source of truth)，而 Elasticsearch/OpenSearch 是您的只读搜索层 (read-only search layer)。
• 您需要将关系数据 (relational data) 非规范化 (denormalize) 为 Elasticsearch/OpenSearch 等 NoSQL 数据源 (NoSQL data source)。
• 您的数据不断变化。
• 您的数据存在于像 Postgres 或 MySQL 或 MariaDB 这样的关系数据库 (relational database) 中，并且您需要一个像 Elasticsearch/OpenSearch 这样的辅助 NoSQL 数据库 (secondary NoSQL database) 进行基于文本的查询 (text-based queries) 或自动补全查询 (autocomplete queries)，以镜像 (mirror) 现有数据，而无需您的应用程序执行双写 (dual writes)。
• 您希望在利用 Elasticsearch/OpenSearch搜索能力 (search capabilities) 的同时，保持现有数据不变，通过公开数据的视图 (exposing a view of your data)，而不损害关系数据 (relational data) 的安全性。
• 或者您只是想公开关系数据的视图 (expose a view of your relational data) 以供搜索之用。

工作原理

PGSync 由 Python 编写（支持 3.9 及更高版本），其堆栈 (stack) 由：Redis/Valkey、Elasticsearch/OpenSearch、Postgres/MySQL/MariaDB 和 SQLAlchemy 组成。

PGSync 利用 Postgres 的逻辑解码 (logical decoding) 功能（在 PostgreSQL 9.4 中引入）来捕获连续的更改事件流 (capture a continuous stream of change events)。 需要在您的 Postgres配置文件 (configuration file) 中通过设置 postgresql.conf 文件中的以下内容来启用 (enabled) 此功能：

wal_level = logical  

您可以选择任何枢纽表 (pivot table) 作为您的文档的根 (root of your document)。

PGSync 的查询构建器 (query builder) 会针对您的模式 (schema) 动态 (dynamically) 构建高级查询 (advanced queries)。

PGSync 通过为数据库中的表创建触发器 (creating triggers for tables) 来处理通知事件 (notification events)，从而在事件驱动模型 (event-driven model) 中运行。

这是 PGSync 唯一一次对您的数据库进行任何更改。

注意 (NOTE): 如果您更改了您的 PGSync 模式配置的结构，建议并且在大多数情况下有必要重建您的 Elasticsearch/OpenSearch 索引。

目前有计划支持零停机迁移 (zero-downtime migrations) 以简化此过程。

快速入门

有几种安装和尝试 PGSync 的方法

• 在 Docker 中运行 (Running in Docker) 是最容易启动和运行的方式。
• 手动配置 (Manual configuration)

Book Demo 示例（需要 DigitalOcean 账户）

在设置过程中填写以下内容：

• ELASTICSEARCH_URL 例如：https://user:pass@os-host:443
• REDIS_URL 例如：rediss://default:pass@host:port/0

在 Docker 中运行（使用 Github 存储库）

使用 Docker 启动所有服务。

$ git checkout https://github.com/toluaina/pgsync  

运行：

$ docker-compose up  

显示 Elasticsearch/OpenSearch 中的内容

$ curl -X GET http://[Elasticsearch/OpenSearch host]:9201/reservations/_search?pretty=true  

在 Docker 中运行（使用镜像存储库）

要使用 Docker 启动所有服务，请按照以下步骤操作：

1. 拉取 Docker 镜像：

$ docker pull toluaina1/pgsync:latest

1. 运行容器：

docker run --rm -it \ -e REDIS_CHECKPOINT=true \ -e REDIS_HOST=<redis_host_address> \ -e PG_URL=postgres://<username>:<password>@<postgres_host>/<database> \ -e ELASTICSEARCH_URL=http://<elasticsearch_host>:9200 \ -v "

环境变量 (Environment variable) 占位符 (placeholders) - 完整列表 在此：

• redis_host_address — Redis/Valkey 服务器的地址（例如，对于本地 Docker 设置，为 host.docker.internal）
• username — PostgreSQL/MySQL/MariaDB 用户名
• password — PostgreSQL/MySQL/MariaDB 密码
• postgres_host — PostgreSQL/MySQL/MariaDB 实例 (instance) 的主机地址 (Host address)（例如，host.docker.internal）
• database — PostgreSQL/MySQL/MariaDB 数据库的名称
• elasticsearch_host — Elasticsearch/OpenSearch 实例 (instance) 的地址（例如，host.docker.internal）

手动配置

Postgres 设置

• 确保数据库用户是超级用户 (superuser)
• 启用逻辑解码 (Enable logical decoding)。您还需要在 postgresql.conf 中设置至少两个参数 (parameters)

wal_level = logical

max_replication_slots = 1

• 为了防止您的服务器日志 (server logs) 增长过大，例如在有成本影响 (cost implication) 的云基础设施 (cloud infrastructure) 上运行时。 您可以可选地 (optionally) 使用 max_slot_wal_keep_size对复制槽大小施加上限 (impose a ceiling on the replication slot size)

max_slot_wal_keep_size = 100GB

MySQL / MariaDB 设置

• 在您的 MySQL / MariaDB 设置中启用二进制日志记录 (Enable binary logging)。
• 您还需要在您的 MySQL / MariaDB 配置 my.cnf 中设置以下参数 (parameters)，然后重新启动数据库服务器 (restart the database server)。

server-id = 1 # 任何非零的唯一 ID (unique ID)

log_bin = mysql-bin

binlog_row_image = FULL # 推荐；如果较旧的 MariaDB 不支持，则省略 (omit)

• 可选的内务管理 (housekeeping): binlog_expire_logs_seconds = 604800 # 7 天
• 您需要创建一个具有 REPLICATION SLAVE 和 REPLICATION CLIENT 权限的复制用户 (create a replication user with REPLICATION SLAVE and REPLICATION CLIENT privileges)

sql CREATE USER 'replicator'@'%' IDENTIFIED WITH mysql_native_password BY 'password'; GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'replicator'@'%'; FLUSH PRIVILEGES;

安装

• 使用 pip 从 pypi安装 (Install) PGSync
  • $ pip install pgsync
• 为您的文档表示 (document representation) 创建 (Create) 一个 schema.json
• 引导数据库 (Bootstrap the database)（仅一次 (one time only)）
  • bootstrap --config schema.json
• 使用以下命令运行程序 (Run the program)
  • pgsync --config schema.json
• 或作为守护进程 (daemon) 运行
  • pgsync --config schema.json -d

特性

PGSync 的关键特性是：

• 轻松地将关系数据非规范化 (Easily denormalize relational data)
• 适用于任何 PostgreSQL 数据库 (PostgreSQL database)（9.6 或更高版本）
• 对数据库性能 (database performance) 的影响可以忽略不计 (Negligible impact)
• Elasticsearch/OpenSearch 中的事务一致性输出 (Transactionally consistent output):
  • 写入 (Writes) 仅在提交 (committed) 后才出现
  • 插入 (Inserts)、更新 (updates) 和删除 (deletes) 以提交顺序 (commit order) 出现（而不是最终 (eventually)）
• 容错 (Fault-tolerant): 即使在崩溃 (crashes) 或网络问题 (network issues) 时也不会丢失数据；处理将从上次的检查点恢复 (processing resumes from the last checkpoint)
• 直接将数据作为 PostgreSQL/MySQL/MariaDB JSON 返回 (Returns data directly as PostgreSQL/MySQL/MariaDB JSON) 以提高速度
• 支持复合主键 (composite primary) 和外键 (foreign keys)
• 支持视图 (views) 和物化视图 (materialized views)
• 处理任意深度的相关表嵌套 (Handles arbitrarily deep nesting of related tables)
• 支持 PostgreSQL/MySQL/MariaDB JSON 字段 (JSON fields)，允许将 JSON 属性 (JSON properties) 作为单独的文档字段 (separate document fields) 提取
• 可定制的文档结构 (Customizable document structure)

要求

• Python 3.9+
• Postgres 9.6+ 或 MySQL 5.7.22+ 或 MariaDB 10.5.0+
• Redis 3.1.0+ 或 Valkey 7.2.0+
• Elasticsearch 6.3.1+ 或 OpenSearch 1.3.7+
• SQLAlchemy 1.3.4+

示例

考虑这个图书库数据库 (Book library database) 的示例。

Book

Author

BookAuthor

使用 PGSync，我们可以简单地定义 (define) 这个 JSON模式 (schema)，其中 book 表是枢纽 (pivot)。 枢纽表 (pivot table) 指示您的文档的根 (indicates the root of your document)。

{  
  "table": "book",  
"columns": [  
    "isbn",  
    "title",  
    "description"
  ],  
"children": [  
    {  
      "table": "author",  
      "columns": [  
        "name"
      ]  
    }  
  ]  
}  

以获取 Elasticsearch/OpenSearch 中的此文档结构 (document structure):

[  
 {  
   "isbn": "9785811243570",  
   "title": "Charlie and the chocolate factory",  
   "description": "Willy Wonka’s famous chocolate factory is opening at last!",  
   "authors": ["Roald Dahl"]  
 },  
 {  
   "isbn": "9788374950978",  
   "title": "Kafka on the Shore",  
   "description": "Kafka on the Shore is a 2002 novel by Japanese author Haruki Murakami",  
   "authors": ["Haruki Murakami", "Philip Gabriel"]  
 },  
 {  
   "isbn": "9781471331435",  
   "title": "1984",  
   "description": "1984 was George Orwell’s chilling prophecy about the dystopian future",  
   "authors": ["George Orwell"]  
 }  
]  

在幕后，PGSync 正在为您生成高级查询 (generating advanced queries)，例如。

SELECT  
   JSON_BUILD_OBJECT(  
     'isbn', book_1.isbn,  
     'title', book_1.title,  
     'description', book_1.description,  
     'authors', anon_1.authors  
   ) AS"JSON_BUILD_OBJECT_1",  
   book_1.id  
FROM book AS book_1  
LEFTOUTERJOIN
 (SELECT
     JSON_AGG(anon_2.anon) ASauthors,  
     book_author_1.book_isbn AS book_isbn  
FROM book_author AS book_author_1  
LEFTOUTERJOIN
  (SELECT
      author_1.name AS anon,  
      author_1.id ASid
   FROM author AS author_1) AS anon_2 ON anon_2.id = book_author_1.author_id  
GROUPBY book_author_1.book_isbn) AS anon_1 ON anon_1.book_isbn = book_1.isbn  

您还可以通过模式配置 (schema config) 配置 (configure) PGSync 来重命名属性 (rename attributes)，例如

 {  
   "isbn": "9781471331435",  
   "this_is_a_custom_title": "1984",  
   "desc": "1984 was George Orwell’s chilling prophecy about the dystopian future",  
   "contributors": ["George Orwell"]  
 }  

PGSync 解决了常见的数据一致性挑战 (common data consistency challenges)，例如：

• 更新数据库中作者的姓名 (Updating an author's name in the database)
• 向现有书籍添加额外的作者 (Adding an additional author to an existing book)
• 更改许多现有文档中作者的姓名 (Changing an author's name across many existing documents)
• 删除或更新作者记录 (Deleting or updating an author record)
• 截断整个表并保持索引同步 (Truncating an entire table and keeping indexes in sync)

优势

PGSync 是一种简单、开箱即用 (simple, out-of-the-box) 的变更数据捕获 (change data capture) 解决方案，旨在最大限度地减少开发工作量 (minimize development effort) 并保持您的搜索索引同步 (keep your search indexes in sync)。

• 自动处理数据删除 (Handles data deletions automatically)。
• 只需要最少的设置 (Requires minimal setup)。只需定义描述您的数据的模式配置 (define a schema config that describes your data)。
• 直接从您的模式生成高级查询 (Generates advanced queries directly from your schema)。
• 使模式更改后重建索引变得容易 (Makes it easy to rebuild indexes after schema changes)。
• 让您只公开您需要的 Elasticsearch/OpenSearch 中的数据 (Lets you expose only the data you need in Elasticsearch/OpenSearch)。
• 支持多租户应用程序的多个 Postgres/MySQL/MariaDB 模式 (Supports multiple Postgres/MySQL/MariaDB schemas for multi-tenant applications)。