使用Go构建一个Postgres流平台

使用 Go 通道从拉推模型转向更高效的流方法。这通过重叠拉取和推送阶段来提高性能，减少总体处理时间和延迟。

Go通道提供数据同步、资源管理和并发处理。它们允许 goroutine 安全地通信和交换数据。这些源实现了每秒 10-12k 事务的吞吐量，最小延迟为 1-5 秒，比之前使用拉推的 30 秒有了显着改进。

在PeerDB，我们的使命是创建一个 Postgres 优先的数据移动平台，使数据从 Postgres 流式传输到数据仓库、队列和存储变得快速、简单。我们的工程重点围绕数据移动速度提高 10 倍、成本效益和硬件优化。

在这篇博文中，我们将深入探讨最近从拉推模型到使用Go goroutine 的更高效流媒体方法的转变。让我们探讨为什么流式传输至关重要，以及这种变化如何显着提高性能。

Pull-and-Push推拉模型
拉推模式：将行提取到内存中的一个数组，然后将它们移动到目标位置。

虽然这种方法在批量较小的情况下效果不错，但在批量较大的情况下就出现了问题。

具体来说，我们无法在拉取的同时并行推送，导致管道效率不高。在我们的典型设置中，拉取和推送时间的比例为 60-40。

_// sync all the records normally, then apply the schema delta after NormalizeFlow._type RecordsWithTableSchemaDelta struct {
RecordBatch            *RecordBatch // wrapper for "Records []Record" TableSchemaDeltas      []*protos.TableSchemaDelta
RelationMessageMapping RelationMessageMapping
}

转向流式处理
我们的新方法是在从 PostgreSQL 提取数据的同时，分批缓冲并并发地将数据推送到目标（如 Snowflake）。这种流水线式数据传输具有显著优势：

• 提高效率：管道化允许我们重叠拉取和推送阶段，从而减少整体处理时间。
• 减少延迟：使用流水线技术，数据可以更快地到达目的地，从而提高整个系统的响应速度。

这是更改后的共享结构：

type CDCRecordStream struct {
// Records are a list of json objects. records chan Record // Schema changes from the slot SchemaDeltas chan *protos.TableSchemaDelta // Relation message mapping RelationMessageMapping chan *RelationMessageMapping // ... other fields
}

利用 Go Channels 进行流式传输
Go Channels 用于实现 Go 程序中 goroutine（并发函数）之间的通信和同步。通道允许一个 goroutine 向另一个 goroutine 发送数据，并提供一种安全的信息交换方式。以下是 Go 通道提供的一些好处：

• 数据同步： Go 通道提供对数据同步的精细控制，防止竞争条件并确保数据流经系统时的一致性。
• 资源管理： Go 通道的满负荷阻塞行为可防止数据过载，降低内存不足 (OOM) 错误的风险并确保稳定性。
• 并发处理： Go 通道可实现高效的并发数据处理，优化资源利用率并在数据检索、转换和插入方面实现高吞吐量。
• 错误处理：使用select 语句的内置错误处理机制提高了系统的健壮性，使我们能够优雅地响应异常并保持可靠性。[这](https://github.com/PeerDB-io/peerdb/blob/57abb885e3e989119bff723340f9d648f5c369bd/flow/connectors/postgres/qrep_query_executor.goL184)是我们在 Go 通道中处理错误的实现
• 与 Postgres 逻辑复制的协同：我们使用逻辑复制槽从 Postgres 管理 CDC。[START_REPLICATION](https://www.postgresql.org/docs/current/protocol-replication.htmlPROTOCOL-REPLICATION-START-REPLICATION)将给定 wal 位置处的 Postgres 更改传输到我们的缓冲区通道中，并等待我们请求下一个更改。Go 通道提供的反压机制和 START_REPLICATION 的流功能齐头并进，通过控制内存利用率来确保弹性。

在最初的规模测试中，我们实现了：

• 吞吐量：每秒 10-12k 事务 (TPS)

• 最小延迟： 1-5 秒

前完成类似任务大约需要 30 秒。

https://www.jdon.com/69634.html