了解如何為 Elasticsearch 創(chuàng)建自定義連接器以簡化數(shù)據(jù)攝取過程。

作者：JEDR BLASZYK

Elasticsearch 擁有一個攝取工具庫，可以從多個來源獲取數(shù)據(jù)。 但是，有時你的數(shù)據(jù)源可能與 Elastic 現(xiàn)有的提取工具不兼容。 在這種情況下，你可能需要創(chuàng)建自定義連接器以將數(shù)據(jù)與 Elasticsearch 連接。

在你的應(yīng)用程序中使用 Elastic 連接器有多種原因。 例如，你可能想要：

• 將數(shù)據(jù)從自定義或遺留應(yīng)用程序引入 Elasticsearch
• 為你的組織數(shù)據(jù)引入語義搜索
• 從 PDF、MS Office 文檔等文件中提取文本內(nèi)容
• 使用 Kibana UI 管理你的數(shù)據(jù)源（包括配置、過濾規(guī)則、設(shè)置定期同步計劃規(guī)則）
• 你想要在自己的基礎(chǔ)設(shè)施上部署 Elastic 連接器（一些 Elastic 支持的連接器可作為 Elastic Cloud 中的本機連接器使用）

用于創(chuàng)建定制連接器的開放代碼框架

如果創(chuàng)建你自己的連接器是滿足你需求的解決方案，那么連接器框架將幫助你創(chuàng)建一個。 我們創(chuàng)建的框架是為了支持創(chuàng)建自定義連接器并幫助用戶將獨特的數(shù)據(jù)源連接到 Elasticsearch。 連接器的代碼可在 GitHub 上找到，并且我們有可以幫助你入門的文檔。

該框架設(shè)計簡單且高性能。 它旨在對開發(fā)人員友好，因此它是開放代碼且高度可定制的。 你創(chuàng)建的連接器可以在你自己的基礎(chǔ)設(shè)施上進行自我管理。 目標(biāo)是讓開發(fā)人員能夠輕松地將自己的數(shù)據(jù)源與 Elasticsearch 集成。

使用連接器框架之前你需要了解什么

該框架是用 async-python 編寫的

有幾門課程可以學(xué)習(xí) async-python。 如果你需要推薦，我們認(rèn)為這個 LinkedIn 學(xué)習(xí)課程非常好，但需要訂閱。 我們喜歡的一個免費替代方案是這個。

為什么我們選擇異步 Python？

攝取受 IO 限制（而非 CPU 限制），因此從資源利用的角度構(gòu)建連接器時，異步編程是最佳方法。 在 I/O 密集型應(yīng)用程序中，大部分時間都花在等待外部資源上，例如讀取文件、發(fā)出網(wǎng)絡(luò)請求或查詢數(shù)據(jù)庫。 在這些等待期間，傳統(tǒng)的同步代碼會阻塞整個程序，導(dǎo)致資源利用效率低下。

還有其他先決條件嗎？

這不是先決條件。 在開始之前，絕對值得閱讀《連接器開發(fā)人員指南》！ 希望你覺得這個有用。

使用連接器框架構(gòu)建定制連接器

入門很容易。 在與框架相關(guān)的術(shù)語中，我們將自定義連接器稱為源。 我們通過創(chuàng)建一個新類來實現(xiàn)一個新的源，該類的職責(zé)是將文檔從自定義數(shù)據(jù)源發(fā)送到Elasticsearch。

作為一種可選的入門方式，用戶還可以查看此目錄源?(directory source) 示例。 這是一個很好但基本的示例，可以幫助你了解如何編寫自定義連接器。

步驟概要

一旦你知道要為其創(chuàng)建連接器的自定義數(shù)據(jù)源，以下是編寫新源的步驟概述：

• 在 connectors/sources 中添加模塊或目錄
• 在 requirements.txt 中聲明你的依賴項。 確保固定這些依賴項
• 實現(xiàn)一個類，該類實現(xiàn) connectors.source.BaseDataSource 中描述的方法
• （可選，在為 repo 做出貢獻時）在 connectors/sources/tests 中添加單元測試，覆蓋率 +90%
• 在源部分聲明你的連接器 connectors/config.py
• 就是這樣。 我們完成了！ 現(xiàn)在你應(yīng)該能夠運行連接器

在編寫定制連接器之前你需要了解什么

為了使 Elasticsearch 用戶能夠獲取數(shù)據(jù)并在該數(shù)據(jù)的基礎(chǔ)上構(gòu)建搜索體驗，我們提供了一個輕量級的連接器協(xié)議。 該協(xié)議允許用戶輕松獲取數(shù)據(jù)、使用企業(yè)搜索功能來操作該數(shù)據(jù)并創(chuàng)建搜索體驗，同時在 Kibana 中為他們提供無縫的用戶體驗。 為了與企業(yè)搜索兼容并充分利用 Kibana 中提供的連接器功能，連接器必須遵守該協(xié)議。

關(guān)于連接器協(xié)議你需要了解的內(nèi)容

該文檔頁面提供了該協(xié)議的良好概述。 以下是你需要了解的內(nèi)容：

• 連接器和系統(tǒng)其他部分之間的所有通信都通過 Elasticsearch 索引異步進行
• 連接器將其狀態(tài)傳達給 Elasticsearch 和 Kibana，以便用戶可以為其提供配置并診斷任何問題
• 這允許簡單、開發(fā)人員友好的連接器部署。 connectors 服務(wù)是無狀態(tài)的，并且不關(guān)心你的 Elastic 部署在哪里運行，只要它可以通過網(wǎng)絡(luò)連接到它就可以正常工作。 該服務(wù)還具有容錯能力，可以在重新啟動或發(fā)生故障后在不同的主機上恢復(fù)操作。 一旦與 Elasticsearch 重新建立連接，它將繼續(xù)正常運行。
• 在底層，該協(xié)議使用 Elasticsearch 索引來跟蹤連接器狀態(tài)
  • .elastic-connectors 和 .elastic-connectors-sync-jobs （在上面鏈接的文檔中描述）

托管自定義連接器的位置

連接器本身不依賴于 Elasticsearch，它可以托管在你自己的環(huán)境中

如果你有 Elasticsearch 部署，無論它是自我管理還是位于 Elastic Cloud 中：

• 作為開發(fā)人員/公司，你可以為你的數(shù)據(jù)源編寫自定義連接器
• 在你自己的基礎(chǔ)設(shè)施上管理連接器并根據(jù)你的需求配置連接器服務(wù)
• 只要連接器可以通過網(wǎng)絡(luò)發(fā)現(xiàn) Elasticsearch，它就能夠?qū)?shù)據(jù)建立索引
• 作為管理員，你可以通過 Kibana 控制連接器

示例：使用連接器框架的 Google Drive 連接器

我們使用連接器框架為 Google Drive 編寫了一個簡單的連接器。 我們通過創(chuàng)建一個新類來實現(xiàn)新的源，該類的職責(zé)是將文檔從目標(biāo)源發(fā)送到 Elasticsearch。

注意：本教程與 Elastic stack 版本 8.10 兼容。 對于更高版本，請務(wù)必檢查連接器發(fā)行說明以獲取更新并參考 Github 存儲庫。

我們從具有 BaseDataSource 預(yù)期方法簽名的 GoogleDriveDataSource 類開始，以配置數(shù)據(jù)源、檢查其可用性（ping）并檢索文檔。 為了使這個連接器發(fā)揮作用，我們需要實現(xiàn)這些方法。

class GoogleDriveDataSource(BaseDataSource):"""Google Drive"""name = "Google Drive"service_type = "google_drive"@classmethoddef get_default_configuration(cls):"""Returns a dict with a default configuration"""raise NotImplementedErrorasync def ping(self):"""When called, pings the backendIf the backend has an issue, raises an exception"""raise NotImplementedErrorasync def get_docs(self, filtering=None):"""Returns an iterator on all documents present in the backendEach document is a tuple with:- a mapping with the data to index- a coroutine to download extra data (attachments)The mapping should have least an `id` fieldand optionally a `timestamp` field in ISO 8601 UTCThe coroutine is called if the document needs to be syncedand has attachments. It needs to return a mapping to index.It has two arguments: doit and timestampIf doit is False, it should return None immediately.If timestamp is provided, it should be used in the mapping.Example:async def get_file(doit=True, timestamp=None):if not doit:returnreturn {'TEXT': 'DATA', 'timestamp': timestamp,'id': 'doc-id'}"""raise NotImplementedError

這個 GoogleDriveDataSource 類是編寫 Google Drive 源代碼的起點。 通過執(zhí)行以下步驟，你將實現(xiàn)與 Google Drive 同步數(shù)據(jù)所需的邏輯：

• 我們需要將此文件添加到 connectors/sources 中
• 設(shè)置新的連接器名稱和 service_type，例如 Google Drive 作為名稱，google_drive 作為服務(wù)類型 (service type)
• 要從源獲取連接器同步數(shù)據(jù)，你需要實現(xiàn)：
  • get_default_configuration - 此函數(shù)應(yīng)返回 RichConfigurableFields 的集合。 這些字段允許你從 Kibana UI 配置連接器。 這包括傳遞身份驗證詳細(xì)信息、憑據(jù)和其他特定于源的設(shè)置。 Kibana 巧妙地呈現(xiàn)這些配置。 例如，如果你將某個字段標(biāo)記為 "sensitive": true，?Kibana 會出于安全原因屏蔽它。
  • ping - 對數(shù)據(jù)源的簡單調(diào)用，驗證其狀態(tài)，將其視為健康檢查。
  • get_docs - 此方法需要實現(xiàn)實際從源獲取數(shù)據(jù)的邏輯。 此函數(shù)應(yīng)返回一個異步迭代器，該迭代器返回一個包含以下內(nèi)容的元組：(document, lazy_download)，其中：
    • document - 是遠(yuǎn)程源中項目的 JSON 表示形式。 （如 name, location, table, author, size 等）
    • lazy_download - 是一個協(xié)程，用于下載框架處理的內(nèi)容提取的對象/附件（例如從 PDF 文檔中提取文本）

BaseDataSource 類中還有其他抽象方法。 請注意，如果你只想支持內(nèi)容同步（例如從谷歌驅(qū)動器獲取所有數(shù)據(jù)），則不需要實現(xiàn)這些方法。 它們指的是其他連接器功能，例如：

• 文檔級安全性（get_access_control、access_control_query）
• 高級過濾規(guī)則（advanced_rules_validators）
• 增量同步 (get_docs_incrementally)
• 將來可能會添加其他功能

我們?nèi)绾尉帉懝俜?Elasticsearch Google Drive 連接器

首先實現(xiàn) BaseDataSource 類所需的方法

我們需要實現(xiàn)方法 get_default_configuration、ping 和 get_docs 以使連接器同步數(shù)據(jù)。 因此，讓我們更深入地了解實現(xiàn)。

首先要考慮的是：如何與Google Drive “對話” 來獲取數(shù)據(jù)？

Google 提供了官方的 python 客戶端，但它是同步的，因此同步內(nèi)容可能會很慢。 我們認(rèn)為更好的選擇是 aiogoogle 庫，它提供了用異步 python 編寫的完整客戶端功能。 一開始這可能并不直觀，但使用異步操作來提高性能非常重要。 因此，在此示例中，我們選擇不使用官方谷歌庫，因為它不支持異步模式。

如果你在異步框架中使用同步或阻塞代碼，可能會對性能產(chǎn)生重大影響。 任何異步框架的核心都是事件循環(huán)。 事件循環(huán)允許通過連續(xù)輪詢已完成的任務(wù)并調(diào)度新任務(wù)來并發(fā)執(zhí)行異步任務(wù)。 如果引入阻塞操作，它將停止循環(huán)的執(zhí)行，從而阻止它管理其他任務(wù)。 這本質(zhì)上否定了異步架構(gòu)提供的并發(fā)優(yōu)勢。

下一個問題是連接器身份驗證

我們將 Google Drive 連接器驗證為服務(wù)帳戶。 有關(guān)身份驗證的更多信息可以在這些連接器文檔頁面中找到。

• 服務(wù)帳戶可以使用密鑰進行身份驗證
• 我們通過 Elasticsearch 中的 Kibana UI 將身份驗證密鑰傳遞給服務(wù)帳戶

讓我們看一下 get_default_configuration 實現(xiàn)，它允許最終用戶傳遞憑證密鑰，該憑證密鑰將存儲在索引中以在同步期間進行身份驗證：

class GoogleDriveDataSource(BaseDataSource):"""Google Drive"""name = "Google Drive"service_type = "google_drive"{...}@classmethoddef get_default_configuration(cls):"""Get the default configuration for Google Drive.Returns:dict: Default configuration."""return {"service_account_credentials": {"display": "textarea","label": "Google Drive service account JSON","sensitive": True,"order": 1,"tooltip": "This connectors authenticates as a service account to synchronize content from Google Drive.","type": "str","value": "",},}

接下來我們來實現(xiàn)一個簡單的 ping 方法

我們將對 google Drive api 進行簡單的調(diào)用，例如 /about 端點。

對于此步驟，我們考慮 GoogleDriveClient 的簡化表示。 我們的主要目標(biāo)是指導(dǎo)你完成連接器創(chuàng)建，因此我們不關(guān)注 Google Drive 客戶端的實現(xiàn)細(xì)節(jié)。 然而，最少的客戶端代碼對于連接器的操作至關(guān)重要，因此我們將依賴 GoogleDriveClient 類表示的偽代碼。

class GoogleDriveClient(GoogleAPIClient):"""A google drive client to handle api calls made to Google Drive API."""{... google drive client implementation}async def ping(self):return await self.api_call(resource="about", method="get", fields="kind")class GoogleDriveDataSource(BaseDataSource):"""Google Drive"""name = "Google Drive"service_type = "google_drive"{...}@cached_propertydef google_drive_client(self):"""Initialize and return the GoogleDriveClientReturns:GoogleDriveClient: An instance of the GoogleDriveClient."""self._validate_service_account_json()json_credentials = json.loads(self.configuration["service_account_credentials"])return GoogleDriveClient(json_credentials=json_credentials)async def ping(self):"""Verify the connection with Google Drive"""try:await self.google_drive_client.ping()self._logger.info("Successfully connected to the Google Drive.")except Exception:self._logger.exception("Error while connecting to the Google Drive.")raise

異步 iterator 從 google drive 返回文件以進行內(nèi)容提取

下一步是編寫 get_docs 異步迭代器，該迭代器將從 Google drive 和協(xié)程返回文件以下載它們以進行內(nèi)容提取。 根據(jù)個人經(jīng)驗，開始將 get_docs 作為一個簡單的獨立 python 腳本來實現(xiàn)并獲取一些數(shù)據(jù)通常更簡單。 一旦 get_docs 代碼正常工作，我們就可以將其移動到數(shù)據(jù)源類。

我們看一下 api 文檔，我們可以：

• 使用文 files/list 端點通過分頁迭代 drive 中的文檔
• 使用 files/get 和 files/export 下載文件（或?qū)?google 文檔導(dǎo)出為特定文件格式）

class GoogleDriveDataSource(BaseDataSource):"""Google Drive"""name = "Google Drive"service_type = "google_drive"{...}async def get_content(self, file, timestamp=None, doit=None):"""Extracts the content from a file file.Args:file (dict): Formatted file document.timestamp (timestamp, optional): Timestamp of file last_modified. Defaults to None.doit (boolean, optional): Boolean value for whether to get content or not. Defaults to None.Returns:dict: Content document with id, timestamp & text"""# Code details have been omitted here for brevity. For a complete implementation,# please refer to the connector code on GitHub.async def get_docs(self, filtering=None):"""Executes the logic to fetch Google Drive objects in an async manner.Args:filtering (optional): Advenced filtering rules. Defaults to None.Yields:dict, partial: dict containing meta-data of the Google Drive objects,partial download content function"""async for files_page in self.google_drive_client.list_files():async for file in self.prepare_files(files_page=files_page):yield file, partial(self.get_content, file)

那么這段代碼中發(fā)生了什么？

• list_files 對驅(qū)動器中的文件進行分頁。
• prepare_files 將文件元數(shù)據(jù)格式化為預(yù)期模式
• get_content 是一個下載文件并對其內(nèi)容進行 Base64 編碼的協(xié)程（內(nèi)容提取的兼容格式）

為了簡潔起見，省略了一些代碼細(xì)節(jié)。 有關(guān)完整的實現(xiàn)，請參閱 GitHub 上的當(dāng)前連接器實現(xiàn)。

讓我們運行連接器！

要將自定義連接器集成到框架中，你需要注冊其實現(xiàn)。 通過在 connectors/config.py 的源部分中添加自定義連接器的條目來執(zhí)行此操作。 對于 Google Drive 示例，添加內(nèi)容將如下所示：

"sources": {...,"google_drive": "connectors.sources.google_drive:GoogleDriveDataSource",...
}

現(xiàn)在在 Kibana 界面中：

• 轉(zhuǎn)到 Search?->?Indices?->?Create a new index?->?Use a Connector
• 選擇 Customized connector（使用自定義連接器時）
• 配置你的連接器。 生成 Elasticsearch API 密鑰和連接器 ID，并按照說明將這些詳細(xì)信息放入 config.yml 中，然后啟動連接器。

此時，Kibana 應(yīng)該檢測到您的連接器！ 安排定期數(shù)據(jù)同步或只需單擊 “Sync” 即可開始完全同步。

連接器可以配置為使用 Elasticsearch 的攝取管道在將數(shù)據(jù)存儲到索引之前對數(shù)據(jù)執(zhí)行轉(zhuǎn)換。 一個常見的用例是通過機器學(xué)習(xí)豐富文檔。 例如，你可以：

• 使用文本嵌入模型分析文本字段，該模型將生成數(shù)據(jù)的密集向量表示
• 運行文本分類以進行情感分析
• 使用命名實體識別 (NER) 從文本中提取關(guān)鍵信息

同步完成后，你的數(shù)據(jù)將在搜索優(yōu)化的 Elasticsearch 索引中可用。 此時，你可以深入構(gòu)建搜索體驗或深入分析。

你想創(chuàng)建并貢獻一個新的連接器嗎？

如果你為可能對 Elasticsearch 社區(qū)有所幫助的源創(chuàng)建自定義連接器，請考慮貢獻它。 以下是使定制連接器成為 Elastic 支持的連接器的升級路徑指南。

貢獻連接器的驗收標(biāo)準(zhǔn)

此外，在開始花一些時間開發(fā)連接器之前，你應(yīng)該創(chuàng)建一個問題并尋求有關(guān)連接器及其將使用哪些庫的一些初步反饋。 一旦你的連接器想法得到一些初步反饋，請確保您的項目滿足一些驗收標(biāo)準(zhǔn)：

• 在 connectors/sources 中添加模塊或目錄
• 實現(xiàn)一個類，該類實現(xiàn) connectors.source.BaseDataSource 中描述的所有方法
• 在 connectors/sources/tests 中添加覆蓋率 +90% 的單元測試
• 在源部分的 connectors/config.py 中聲明你的連接器
• 在 requirements.txt 中聲明你的依賴項。 確保固定這些依賴項
• 對于你要添加的每個依賴項（包括間接依賴項），列出所有許可證并在補丁中提供該列表。
• 確保你的源使用異步庫。 如果不可能，請確保沒有阻塞循環(huán)
• 如果可能，請?zhí)峁┻\行后端服務(wù)的 docker 映像，以便我們測試連接器。 如果您無法提供 Docker 映像，請?zhí)峁┽槍υ诰€服務(wù)運行所需的憑據(jù)。
• 由于 Elasticsearch 分頁的默認(rèn)大小限制為 10k，測試后端需要返回超過 10k 的文檔。 從測試后端返回超過 10k 文檔將有助于測試連接器

用于測試連接器的支持工具

我們還有一些支持工具來分析連接器代碼并運行性能測試。 你可以在這里找到這些資源：

• Perf8 - 性能庫和儀表板，用于分析 python 代碼的質(zhì)量，以評估資源利用率并檢測阻塞調(diào)用
• E-2-E 功能測試，利用 perf8 庫來分析每個連接器

總結(jié)

我們希望這個博客和示例對你有用。

以下是 Elasticsearch 可用的 native connectors 和 connector clients?的完整列表。 如果你沒有找到列出的數(shù)據(jù)源，是否可以創(chuàng)建一個自定義連接器？

以下是與本文相關(guān)的一些有用資源：

• 連接器 GitHub 存儲庫和文檔頁面
• 異步 Python 學(xué)習(xí)課程
• 新的自定義連接器社區(qū)指南
• Elastic 連接器框架的許可詳細(xì)信息（在此鏈接中搜索 'Connector Framework'）

如果你沒有 Elastic 帳戶，你可以隨時啟動試用帳戶來開始！