Elastic Connectors：增量同步对性能的影响

作者：Artem Shelkovnikov

Elastic 连接器是一种 Elastic 集成，可将数据从原始数据源同步到 Elasticsearch 索引。连接器使你能够创建可搜索的只读数据源副本。

有许多连接器支持各种第三方，例如：

MongoDB
各种 SQL DBMS，例如 MySQL、PostgreSQL、MSSQL 和 OracleDB
Sharepoint Online
Amazon S3
还有更多。完整列表可在此处查看。

连接器支持两种类型的内容同步作业：完全同步和增量同步。

完全同步

完全同步是从第三方服务中提取所有所需文档并将其导入 Elasticsearch 的同步。因此，如果你已将网络驱动器连接器设置为从文件夹 “\Documents/Reports\2022**.docx” 导入所有文档，则在完全同步期间，连接器将获取符合此条件的所有文档并将其全部发送到 Elasticsearch。简化的伪代码如下所示：

connector = NetworkDriveConnector(host="192.168.0.105",path="\\Documents\Reports\2025\**.doc"
)for incoming_document_metadata in connector.extract_documents():content = connector.download(incoming_document_metadata)document = {"id": incoming_document_metadata["id"]"content": content}elasticsearch.ingest(document)

这种方法效果很好，直到同步开始耗时过长。这可能是因为连接器获取了比所需更多的数据。例如，为什么要获取未更改的旧文件并将其发送到 Elasticsearch？有人可能会说文件的元数据可能不可靠，因此需要再次获取所有文件并将其发送到 Elasticsearch。确实，情况可能如此，但如果我们可以信任从第三方获取的数据的元数据，我们就可以提取更少的数据。增量同步是实现此目的的方法。

增量同步

大多数情况下，如果编写得当，连接器会花大量时间执行 IO 操作。回到示例代码，有 3 个地方会发生 IO：

connector = NetworkDriveConnector(host="192.168.0.105",path="\\Documents\Reports\2025\**.doc"
)# Place #1: reading document metadata from 3rd-party system
for incoming_document_metadata in connector.extract_documents():# Place #2: reading document content from 3rd-party systemcontent = connector.download(incoming_document_metadata)document = {"id": incoming_document_metadata["id"]"content": content}# Place #3: ingesting the resulting document into Elasticsearchelasticsearch.ingest(document)

这些地方都可能成为瓶颈，并在同步过程中耗费大量时间。

这就是增量同步发挥作用的地方。其目的是尽可能减少任何阶段的 IO 量。

潜在的优化

从第三方系统获取更少的文档

修改上面的示例，代码可能如下所示：

connector = NetworkDriveConnector(host="192.168.0.105",path="\\Documents\Reports\2025\**.doc"
)# We can store last sync time somewhere
last_sync_time = connector.fetch_last_sync_time()# And later use it querying Network Drive
for incoming_document_metadata in connector.extract_documents(from=last_sync_time):content = connector.download(incoming_document_metadata)document = {"id": incoming_document_metadata["id"]"content": content}elasticsearch.ingest(document)

如果我们的第三方系统中只有少量文档发生变化，我们可以显著加快提取过程。但是，对于网络驱动器（netword drive）来说，这是不可能的 - 它的 API 不支持通过元数据过滤文档。我们将无法避免扫描网络驱动器的全部内容。

跳过下载自上次同步以来未发生更改的文件内容

在同步过程中，下载文件内容需要花费大量时间。如果文件相当大、连接不稳定或吞吐量较低，则在同步第三方内容时，下载文件内容将花费大部分时间。如果我们跳过下载其中一些文件，这已经可以显著加快连接器的速度。

考虑以下示例伪代码：

connector = NetworkDriveConnector(host="192.168.0.105",path="\\Documents\Reports\2025\**.doc"
)last_sync_time = connector.fetch_last_sync_time()for incoming_document_metadata in connector.extract_documents():# If document timestamp did not change then not fetching# document content can save us a lot of timeif incoming_document_metadata["last_updated_at"] > last_sync_timecontent = connector.download(incoming_document_metadata)document = {"id": incoming_document_metadata["id"]"content": content}elasticsearch.ingest(document)

如果没有更新文档，同步速度实际上会比完全同步内容快几个数量级。

跳过将未修改的文档提取到 Elasticsearch 的过程

虽然这看起来微不足道，但将数据提取到 Elasticsearch 需要花费大量时间 - 尽管通常比从第三方系统下载内容的时间要短。我们可以开始存储每个文档的时间戳，如果时间戳没有改变，则不将文档发送到 Elasticsearch。

我们可以将此方法与之前的方法结合起来，以在同步期间节省尽可能多的时间。

connector = NetworkDriveConnector(host="192.168.0.105",path="\\Documents\Reports\2025\**.doc"
)# We need to fetch only IDs and timestamps as it's sufficient to make a decision.
# For large indices it can still take a good amount of RAM, but that's the price.
existing_documents = connector.fetch_existing_documents(fields=["id", "_timestamp"])for incoming_document_metadata in connector.extract_documents():existing_document_metadata = existing_documents[document_metadata["id"]]# If a document for this 3rd-party record exists in Elasticsearch index# and timestamp did not change, then skip downloading its content# and skip ingesting the documentif existing_document_metadata:incoming_document_timestamp = incoming_document_metadata["last_updated_at"]existing_document_timestamp = existing_document_metadata["_timestamp"]if incoming_document_timestamp == existing_document_timestamp:# Skip the document for goodcontinue;content = connector.download(incoming_document_metadata)document = {"id": incoming_document_metadata["id"]"content": content,"_timestamp" = incoming_document_metadata["last_updated_at"]}elasticsearch.ingest(document)

这种方法有助于在运行同步时节省更多时间。现在让我们看看此类改进的性能考虑因素。

增量同步性能

现在，既然我们已经研究了简化的代码，了解了增量同步的工作原理，我们可以尝试估计潜在的性能改进。

对于某些连接器，增量同步以某种方式实现，以优化从第三方获取数据的方式。例如，Sharepoint Online 连接器通过增量 API 获取一些数据 - 仅收集上次同步后更改的文档。这以明显的方式提高了性能 - 更少的数据 -> 更少的时间将数据同步到最新状态。

对于其他连接器（目前除 Sharepoint Online 连接器外的所有连接器），增量同步由框架以通用方式完成，这在前面的章节之一 “跳过将未修改的文档提取到 Elasticsearch” 中进行了描述。

连接器仍然从第三方数据源收集所有数据（因为它不提供仅提取更改的记录的方法）。但是，如果此数据包含时间戳，则连接器框架会将已提取文档的文档 ID 和时间戳与传入文档进行比较。如果 Elasticsearch 中存在文档，并且时间戳与从第三方数据源收到的时间戳相同，则不会将此文档发送到 Elasticsearch。

我们已经描述了使用增量同步来提高性能的抽象方法，但我们已经在连接器中实现了这些方法，所以让我们深入研究一下实际数字吧！

性能测试

我们将通过这些性能测试估计增量同步的粗略改进幅度，但目标不是高精度。

之所以选择 Google Drive 和 Github 作为本次测试的两个连接器，是因为它们具有不同的 IO 配置文件。

Google Drive 就像一个文件存储。它：

有一个快速的 API，不会过早节流
通常存储大量可变大小的二进制内容 - 从小到非常大
通常存储少量记录 - 数万或数十万而不是数百万

GitHub 数据是通过更经典的 API 提取的，它：

经常节流
包含许多比 Google Drive 中的记录小得多的记录
根本不发送二进制内容

由于这些差异，增量同步性能将有很大不同。

两项测试都包含以下强制性步骤：

对第三方系统进行完全同步
修改第三方系统上的某些文档
运行增量同步并检查所需的时间

此设置非常简单，但已经可以很好地表明性能改进的程度。两项测试略有不同，我将在下一节中提供结果和评论。

设置 #1 - Google Drive 连接器

初始设置如下：

Google Drive 上有 1 个文件夹，其中包含 1553 个文件（其中 100 个大小为 2MB，1443 个大小为 5KB）
执行完全同步并将这些数据输入 Elasticsearch
将更多文件添加到 Google Drive 中，使其文件数达到 10144 个（其中 100 个大小为 2MB，其余文件大小均为 5KB）
再次执行增量同步以提取新数据
然后对 Google Drive 上的文件进行一些细微更改（添加 1 个，删除 2 个）
再次执行增量同步
再次执行完全同步以将运行时间与增量同步进行比较

下表包含所述测试的结果和注释：

Sync 描述	运行时间	添加文档数	文档删除数	注释
初始完全同步	0h 4m 0s	1553	0	这是初始同步 - 它会提取所有文档
在 Google Drive 中添加更多数据后进行增量同步	0h 20m 9s	7939	0	运行时间如预期一样长 - 大量文档输入
Google Drive 中的部分数据发生轻微更改后进行增量同步	0h 1m 25s	1	2	运行速度非常快。它仍然大量调用 Google Drive API，但不必将 200+MB 的数据导入 Elasticsearch
完全同步以比较性能	0h 23m 23s	10144	0	正如预期的那样，这需要花费大量时间 —— 所有数据都从 Google Drive 下载并发送到 Elasticsearch，即使数据没有变化。我们可以假设下载数据并将其上传到 Elasticsearch 进行设置需要 22 分钟

总之，增量同步显著提高了连接器的性能，因为大部分时间都花在连接器下载文件内容并将这些内容发送到 Elasticsearch 上。完全同步带来 2 * 100 + 1443 * 5 / 1024 = 207MB 的内容 - 既由连接器下载，又被提取到 Elasticsearch 中。如果只更改 1 个大文件，则这个数量只会变为 2MB - 变化幅度为 100。这很好地解释了性能改进。

设置 #2：GitHub 连接器

GitHub 连接器非常不同，因为它同步的实际数据量相对较小 - 问题、拉取请求等相当小，而它们却很多。此外，GitHub 有严格的限制政策，并且会大量限制连接器。

为了给出一个很好的现实世界示例，我们将使用 Kibana Github 存储库和 GitHub 连接器并观察其性能。

Sync 描述	运行时间	添加的文档数	删除的文档数	注释
初始完全同步	8h 40m 1s	147421	0	---
增量同步立即运行	9h 6m 7s	59	0	这次同步花费了更多时间，主要是因为它不断受到限制。连接器必须从 GitHub 获取所有数据，但只发送了 59 条记录，总容量不到 1MB
下一次增量同步	9h 2m 52s	191	1	本次同步是在上一次增量同步完成后立即触发的。由于数据几乎相同，且节流是连接器运行时间的主要因素，因此运行时间相同

关键要点

如你所见，增量同步对 Github 连接器的性能没有任何改进 - 几乎没有任何优化空间，因为连接器大部分时间都花在查询系统和等待节流停止上。
提取的文档相当小，因此网络吞吐量使用量很小。为了缩短连接器的运行时间，增量同步实际上必须限制对 Github 的查询次数，但目前它尚未在连接器中实现。

总结

影响增量同步性能的主要因素是什么？简而言之，就是提取的原始数据量。

对于 Sharepoint Online 连接器，有一种特殊的逻辑可以通过增量 API 获取较少的数据。这节省了大量时间，因为增量 API 允许连接器不提取未更改的文件。文件往往很大，因此不下载和提取它们将节省大量时间。

对于其他连接器，增量同步是通用的 - 它只是在将文档提取到 Elasticsearch 之前检查文档时间戳 - 如果此文档已在索引中并且时间戳没有更改，则不会提取它。它比 Sharepoint Online 采用的以前的方法节省的时间更少，但适用于所有连接器。一些连接器（包含大型文档的连接器）从这种逻辑中受益匪浅，而其他连接器（受到第三方系统限制且包含相对较小的文档的连接器）则无法从增量同步中获益。

此外，如果 Elasticsearch 负载过重，增量同步不太可能受到 Elasticsearch 的限制，从而使其在负载下性能更高。

让我们看看下面的图表：