node.js - 将 elasticsearch & cassandra 与 postgres 数据库同步

Question

我想将两个依赖数据库(elasticsearch 和 casandra)与我的父数据库同步:postgres。我正在尝试实现本文中的一种方法:https://qafoo.com/blog/086_how_to_synchronize_a_database_with_elastic_search.html .所以我想出了2种方法

1. 在将数据更新/插入相关数据库之前进行同步

    router.put('/account/edit', function(req, res) { 
   
     syncElasticWithDatabase().then(() => { 
            elastiClient.update({...}); // client for elasticsearch
            cassandraClient.execute({...}); // client for cassandra
             req.end();
     })
   
    })
   

syncElasticWithDatabase() 使用 updates 表中的数据(来自 postgres)，此方法可能很慢，因为有些人必须等待 syncElasticWithDatabase() 完成。我喜欢这种方法，因为我利用了 sequantial_ids(查看文章了解详情)。在新数据进入之前同步数据，允许依赖项 catch 并且只同步丢失的数据。与下面的选项 2 不同，防止重建索引/重新插入。

1. 使用后台进程(ei:每 24 小时运行一次)，我可以通过从 update_error 表中选择“丢失的数据”来同步数据，该表包含 elasticsearch 或 cassandra 失败时的数据。这是一个粗略的例子

    router.put('/account/edit', function(req, res) { 
   
     psqlClient.query('UPDATE....').then(() => { 
          elastiClient.update({...}); // client for elasticsearch
          cassandraClient.execute({...}); // client for cassandra
     }).catch(err => {
         psqlClient.query('INERT INTO update_error ....')
         })
   
     })
   

   然而，这种方法需要重新索引或重新插入数据，因为在某些情况下，elasticsearch 可以插入数据，而 cassandra 则不能，或者以任何一种方式插入。因此，我需要一个单独的列来记录失败的数据库类型。这样我就可以为每种类型的数据库(ealsticsearch 或 cassandra)选择自上次同步时间以来失败的数据。

问题:

1. 方法 1 似乎很完美，但这意味着由于 syncElasticWithDatabase()，有些人必须比其他人等待更长时间才能更新他们的帐户。但是上面的文章完全一样(看他们的图表)还是我误解了什么？

2. 由于上述延迟(如果我是正确的)，我引入了选项 2。但是，为了同步恕我直言，它太多了。然而我花了很长时间思考这个问题……那么有没有比 1 和 2 更简单或更好的方法呢？

3. Apache Zoo Keeper 对我有帮助吗？

谢谢:)

---

其他引用

Sync elasticsearch on connection with database - nodeJS

https://gocardless.com/blog/syncing-postgres-to-elasticsearch-lessons-learned/

Answer 1

基本上，您需要使用此处描述的方法 https://qafoo.com/blog/086_how_to_synchronize_a_database_with_elastic_search.html并从一个数据库表中插入和选择数据。但是请确保在“更新”中选择数据时限制选择的数量，例如:LIMIT 100。

工作流程如下:

1. 在 insert/update 期间将数据保存到“updates”表(如果 delete 确保在列中将其标记为已删除)

2. 然后运行这个过程 ->

   • 选择您最后插入的内容:来自 elasticsearch 或 Cassandra 的 sequence_id
   • 使用它从“更新”表中选择数据，如下所示:id > :sequence_id

然后您可以插入数据(到 elasticsearch 或 cassandra)或做任何事情。确保将数据插入“更新”表 之前 相关数据库。并且不需要复制 document_id 所以用新的替换它们。这提供了一致性，并允许您在运行 cron 作业或在特定操作期间同时同步它之间进行选择。然后将您的 sequence_id 更新为最后一个。

我选择在插入/更新/删除后直接同步数据到“更新”，然后我执行 res.end() (或任何完成响应的操作)并使用 sync() 函数以升序选择 100 条新记录。我还每 24 小时运行一次 cron 作业(没有 LIMIT 100)以确保遗漏的任何数据都将被同步。哦，是的，如果所有数据库的更新都成功，那么您最好从“更新”中删除记录，除非您出于其他原因使用它。但请注意，elasticsearch 可能会丢失内存中的数据

祝你好运 :) 我乐于接受建议