使用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 语句的内置错误处理机制提高了系统的健壮性，使我们能够优雅地响应异常并保持可靠性。这是我们在 Go 通道中处理错误的实现
  • 与 Postgres 逻辑复制的协同：我们使用逻辑复制槽从 Postgres 管理 CDC。START_REPLICATION将给定 wal 位置处的 Postgres 更改传输到我们的缓冲区通道中，并等待我们请求下一个更改。Go 通道提供的反压机制和 START_REPLICATION 的流功能齐头并进，通过控制内存利用率来确保弹性。

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

  • 吞吐量：每秒 10-12k 事务 (TPS)
  • 最小延迟： 1-5 秒

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