python - 持续处理来自 PostGres 数据库的数据——采用什么方法？

Question

对我尝试构建的流程采用哪种方法有疑问。使用 PostgreSQL 和 Python。

场景:

• 我有两个数据库 A 和 B。
• B 是 A 的处理版本。
• 数据源源不断地流入A，需要按一定的时间处理 方式(使用多处理)，然后存储在 B 中。
• A 中的每个新行只需要处理一次。

所以:

streamofdata ===> [数据库A] ----> 进程----> [数据库B]

数据库 A 相当大 (40 GB) 并且还在增长。我的问题是关于确定哪些新数据尚未处理并放入 B。确定哪些行仍需处理的最佳方法是什么。

我猜每次都在尚未处理的内容上匹配主键不是可行的方法

假设新行 120 到 130 在一段时间内进入数据库 A。所以我最后一行处理的行是 119。查看最后一行 id(主键)119 处理并说现在应该处理任何超出该行的正确方法是正确的方法吗？

还想知道是否有人对这种“实时”数据处理有任何进一步的资源。从技术上讲，不太确定我在寻找什么。

Answer 1

好吧，有几种方法可以解决这个问题。提醒一下，您描述的过程基本上是重新实现一种数据库复制形式，因此您可能希望自己熟悉 Postgres 的各种流行复制选项及其工作方式，尤其是 Slony你可能会感兴趣。您没有指定“数据库 B”是什么类型的数据库，所以我假设它是一个单独的 PostgreSQL 实例，尽管除了排除一些固定解决方案(如 Slony)之外，这个假设不会改变下面的决定。

1. 设置 FOR EACH ROW trigger在数据库 A 中需要复制的重要表上。您的触发器将在这些表中插入每个新行(和/或更新，删除，如果您需要捕获它们)并将它们适本地发送到数据库 B。你提到使用 Python，所以提醒你当然可以在 PL/python 中编写这些触发函数。如果这让您的生活变得轻松，即您应该能够或多或少地轻松调整现有代码，使其作为 PL/Python 触发器函数在数据库中运行。

2. 如果您阅读过 Slony，您可能已经注意到提案 #1 与 Slony 的工作方式非常相似——考虑让 Slony 接管必要的复制对您来说是否容易或有帮助从数据库 A 到数据库 B 的表，然后如果您需要进一步将数据移动/转换到数据库 B 中的其他表，您可以在数据库 B 中的这些表上使用触发器。

3. 设置触发器或 RULE这将发出 NOTIFY带有指示已更改行的有效负载。您的代码将 LISTEN对于这些通知并立即知道哪些行已更改。 psycopg2 适配器有 good support听和通知。注意您需要谨慎处理监听器代码崩溃或与数据库断开连接或以其他方式错过某些通知的情况。

4. 如果您可以控制将数据流式传输到数据库 A 的代码，您可以让该代码接管将其新数据复制到数据库 B 的工作。