将 elasticsearch 和 cassandra 与 postgres 数据库同步答案

【问题标题】：Sync elasticsearch & cassandra with postgres database将 elasticsearch 和 cassandra 与 postgres 数据库同步
【发布时间】：2016-10-05 22:00:08
【问题描述】：

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

在更新/插入数据到依赖数据库之前同步

 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（查看文章了解详细信息）。数据在新数据进入之前同步，允许依赖关系赶上，只有错过的数据才会被同步。与下面的选项 2 不同，防止重新索引/重新插入。

使用后台进程（ei：每 24 小时运行一次），我可以通过从 update_error 表中选择“丢失的数据”来同步数据，该表包含弹性搜索或 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 看起来很完美，但这意味着由于syncElasticWithDatabase()，某些人将不得不比其他人等待更长的时间来更新他们的帐户。但是上面的文章完全一样（看他们的图表）还是我误解了什么？
由于上述延迟（如果我是正确的），我介绍了选项 2。但是，为了同步恕我直言，它太多了。然而我花了很多时间思考这个......那么有比1和2更简单或更好的方法吗？
Apache Zoo Keeper 对我有帮助吗？

谢谢:)

其他参考

Sync elasticsearch on connection with database - nodeJS

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

【问题讨论】：

我们如何在 windows 上批量导入 postgres 表到 elasticsearch ！

标签： node.js postgresql elasticsearch cassandra synchronization

【解决方案1】：

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

这是工作流程：

在insert/update 期间将数据保存到“更新”表（如果delete 确保在列中将其标记为已删除）
然后运行这个过程 ->
- 选择你最后一次插入：sequence_id from elasticsearch 或 Cassandra
- 使用它从“更新”表中选择数据，如下所示：id > :sequence_id

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

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

祝你好运 :) 我愿意接受建议

【讨论】：