将数据从 SQL 服务器加载到 Elasticsearch 的模式

Question

这是我们想出的。通过使用 3 值状态列。

0 = Not indexed
1 = Updated
2 = Indexed

将有 2 个工作......

Job 1 将选择 status = 0 的前 X 条记录，并将它们弹出到 RabitMQ 等队列中。 然后消费者会将这些记录批量插入到 ES 中，并将 DB 记录的状态更新为 1。

对于更新，因为我们可以控制我们的数据...更新该特定记录的 SQL 存储过程会将其状态设置为 2。Job2 将选择状态 = 2 的前 x 条记录并将它们弹出到 RabitMQ。然后消费者会将这些记录批量插入到 ES 中，并将 DB 记录的状态更新为 1。

当然，我们可能需要“排队”的中间状态，这样所有作业都不会再次获取相同的记录，但如果尚未完成，则不应运行相同的作业。排队记录被更新的机会微乎其微。因为更新只在一天结束时发生，通常是第二天。

所以我知道有河流（但已被弃用并且可能不像 ETL 那样灵活）

我想将我的 SQL 服务器中的记录批量插入到 Elasticsearch。

编写某种预定的批处理作业，无论是 ETL 还是任何其他工具都无所谓。

从 id > lastIdInsertedToElasticSearch 的表中选择这将允许以预定的时间间隔将最新记录加载到 Elasticsearch 中。

但是如果在 SQL 服务器中更新了一条记录呢？什么是跟踪 SQL 服务器中更新记录然后在 ES 中推送更新记录的好模式？我知道 ES 在放置相同的 ID 时有文档版本。但似乎无法可视化模式。

Answer 1

所以恕我直言，批量插入非常适合构建或重新构建索引。因此，您第一次可以运行运行 SQL 查询和执行批量更新的批处理作业。正如您正确指出的那样，Rivers 在转换方面没有提供很大的灵活性。

如果您的 SQL 数据存储中的条目是由您创建的（即您控制的某些代码库），那么相同的代码库更新 Elasticsearch 中的文档会更好，可能不是直接更新，而是通过通知其他服务或使用队列有助于不浪费时间响应请求（如果您有这种设置）。

我们有一个非常相似的 Elasticsearch 用例。我们在我们的应用程序中提供搜索功能，该应用程序对不同类别的数据进行搜索。其中一些数据实际上是由我们应用程序的用户通过我们的应用程序创建的——因此我们可以轻松处理。我们的应用程序将该数据写入我们的 SQL 数据存储，并将相同的数据推送到 RabbitMQ 中，以便在 Elasticsearch 中进行索引/更新。在 RabbitMQ 的另一端，我们有一个用 Python 编写的消费者，它基本上替换了 Elasticsearch 中的整个文档。因此，我们的 SQL 数据存储中的相应行和 Elasticsearch 中的文档共享 ID，这使我们能够更新文档。

另一种情况是，我们执行搜索的一些类型的数据来自某些通过其 HTTP API 公开数据的第三方服务。数据创建由我们控制，但我们没有自动更新 Elasticsearch 中的条目的机制。在这种情况下，我们基本上运行一个 cron 作业来处理这个问题。我们已经设法调整了 cron 的时间表，因为我们也有有限数量的 API 查询配额。但在这种情况下，我们的数据并没有真正每天更新这么多。所以这种系统适合我们。

Answer 2

免责声明：我共同开发了这个解决方案。

我需要像jdbc-river 这样可以做更复杂的数据“汇总”的东西。在仔细考虑了修改 jdbc-river 以满足我的需要之后，我最终写了river-net。

以下是一些功能：

• 它获得了相当不错的性能（与 jdbc-river 相当。我们获得了超过 6k 行/秒）
• 它可以连接多个表以创建复杂的嵌套文档数组，而无需创建重复的子文档
• 它遵循许多与 jdbc-river 相同的约定。
• 它还支持从文件中读取。
• 它是用 C# 编写的
• 它使用Quartz.Net 并支持 cron 表达式进行调度。

这个项目是开源的，我们已经有第二个项目（也将是开源的）使用RabbitMQ 进行通用作业调度。我们已经移植了很多这个项目，并计划到 RabbitMQ 河中，以便在索引到 Elasticsearch 时获得更好的性能和稳定性。

为了应对大型更新，我们没有直接点击表格。相反，我们使用仅获取增量的stored procedures。我们还可以在 sp 上选择重置增量以重新索引所有内容。

这个项目还很年轻，只有几个提交，但我们愿意合作和新想法。