作者：來自 Elastic?Dmitrii Burlutskii

在 Elastic 的搜索團(tuán)隊中，我們一直在探索不同的 ETL 工具以及如何利用它們將數(shù)據(jù)傳輸?shù)?Elasticsearch，并在傳輸?shù)臄?shù)據(jù)上實(shí)現(xiàn) AI 助力搜索。今天，我想與大家分享我們與 Meltano 生態(tài)系統(tǒng)以及 Meltano Elasticsearch 加載器的故事。

Meltano 是一個聲明式的代碼優(yōu)先數(shù)據(jù)集成引擎，允許你在不同的存儲之間同步數(shù)據(jù)。在 hub.meltano.com 上有許提取器 (extractors) 和加載器 (loaders) 可用。如果你的數(shù)據(jù)存儲在 Snowflake 中，并且想要為你的客戶構(gòu)建一個開箱即用的搜索體驗(yàn)，你可能會考慮使用 Elasticsearch，在那里你可以基于你擁有的數(shù)據(jù)為客戶構(gòu)建語義搜索。今天，我們將重點(diǎn)介紹如何將數(shù)據(jù)從 Snowflake 同步到 Elasticsearch。

要求

Snowflake 賬號。?你在注冊后將收到以下所有賬號信息，或者你可以從 Snowflake 面板中獲取它們。

1. 賬戶用戶名
2. 賬戶密碼
3. 賬戶標(biāo)識符（查看此處的說明以獲取它）

Snowflake 數(shù)據(jù)集

如果你創(chuàng)建了一個新的 Snowflake 賬戶，你將擁有用于實(shí)驗(yàn)的示例數(shù)據(jù)。

然而，我將使用一個公共空氣質(zhì)量數(shù)據(jù)集，其中包含二氧化氮（NO2）的測量數(shù)據(jù)。

Elastic 賬號

訪問 https://cloud.elastic.co 并注冊賬號。

點(diǎn)擊 “Create deployment”。在彈出窗口中，你可以更改或保留默認(rèn)設(shè)置。

一旦準(zhǔn)備好部署，請點(diǎn)擊 “Continue”（或點(diǎn)擊 “Open Kibana”）。它將重定向你到 Kibana 儀表板。

轉(zhuǎn)到 Stack Management?->?Security?->?API keys，并生成一個新的 API 密鑰。

安裝 Meltano

在我的示例中，我將使用 Meltano Python 包，但你也可以將其作為 Docker 容器安裝。

pip install "meltano"

添加 Snowflake 提取器

meltano add extractor tap-snowflake --variant=meltanolabs

驗(yàn)證提取器

meltano invoke tap-snowflake --test

添加 Elasticsearch 加載器

meltano add loader target-elasticsearch

配置提取器和加載器：

有多種方法可以配置 Meltano 提取器和加載器：

• 編輯 meltano.yml
• 使用 CLI 命令，例如

meltano config {loader} set config_name config_value

使用 CLI 交互模式

meltano config {loader} set --interactive

我將使用交互模式。

要配置 Snowflake 提取器，請運(yùn)行以下命令并至少提??供帳戶標(biāo)識符、用戶名、密碼和數(shù)據(jù)庫。

meltano config tap-snowflake set --interactive

你應(yīng)該會看到以下屏幕，你可以在其中選擇要配置的選項。

配置提取后，你可以測試連接。 只需運(yùn)行以下命令：

配置 Elasticsearch 加載器并提供主機(jī)、端口、架構(gòu)和 API 密鑰，

meltano config target-elasticsearch set --interactive

如果你想更改索引名稱，可以運(yùn)行以下命令并更改它：

meltano config target-elasticsearch set index_format my-index-name

meltano config target-elasticsearch set index_format my-index-name

比如，?默認(rèn)索引字符串定義為 ecs-{{ stream_name }}-{{ current_timestamp_daily}} ，結(jié)果為 ecs-animals-2022-12-25，其中流名稱為 animals。

配置完所有內(nèi)容后，我們就可以開始同步數(shù)據(jù)。

meltano run tap-snowflake target-elasticsearch

同步開始后，你可以轉(zhuǎn)到 Kibana 并看到創(chuàng)建了一個新索引并且有一些索引文檔。

你可以通過單擊索引名稱來查看文檔。 你應(yīng)該查看你的文件。

使用你的索引設(shè)置（或映射）

如果我們開始同步數(shù)據(jù)，加載器將自動創(chuàng)建一個具有動態(tài)映射的新索引，這意味著 Elasticsearch 將處理索引中的字段及其類型。 如果我們愿意，我們可以通過提前創(chuàng)建索引并應(yīng)用我們需要的設(shè)置來更改此行為。 咱們試試吧。

導(dǎo)航到 Kibana -> DevTools 并運(yùn)行以下命令：

創(chuàng)建新的攝入管道

PUT _ingest/pipeline/drop-values-10
{"processors": [{"drop": {"description": "Drop documents with the value < 10","if": "ctx.datavalue < 10"}}]
}

這將刪除 datavalue < 10 的所有文檔。

創(chuàng)建新索引

PUT my-snowflake-data

應(yīng)用索引設(shè)置

PUT my-snowflake-data/_settings
{"index": {"default_pipeline": "_ingest/pipeline/drop-values-10"}
}

更改 Meltano 中的索引名稱

meltano config target-elasticsearch set index_format my-snowflake-data

開始同步作業(yè)

meltano run tap-snowflake target-elasticsearch

工作完成后，你可以看到索引中的文檔比我們之前創(chuàng)建的要少

結(jié)論

我們已經(jīng)成功地將數(shù)據(jù)從 Snowflake 同步到 Elastic Cloud。我們讓 Meltano 為我們創(chuàng)建了一個新索引，并負(fù)責(zé)索引映射，我們將數(shù)據(jù)同步到了一個具有預(yù)定義管道的現(xiàn)有索引中。

我想強(qiáng)調(diào)在我旅程中記下的一些關(guān)鍵點(diǎn)：

Elasticsearch 加載器（Meltano Hub 上的頁面）

• 它尚未準(zhǔn)備好處理大量的數(shù)據(jù)。你需要調(diào)整默認(rèn)的 Elasticsearch 配置，使其更加健壯。我已經(jīng)提交了一個 Pull Request，以暴露 “request_timeout” 和 “retry_on_timeout” 選項，這將會有所幫助。
• 它使用 Elasticsearch Python 客戶端的 8.x 分支，因此你可以確保它支持最新的 Elasticsearch 功能。
• 它同步發(fā)送數(shù)據(jù)（不使用 Python AsyncIO），因此當(dāng)您需要傳輸大量數(shù)據(jù)時可能會相當(dāng)慢。

Meltano CLI

• 它非常棒。你不需要 UI，所以一切都可以在終端中配置，這為工程師提供了大量的自動化選項。
• 你可以僅通過一個命令即可運(yùn)行按需同步。不需要其他正在運(yùn)行的服務(wù)。

復(fù)制/增量同步

• 如果你的管道需要數(shù)據(jù)復(fù)制或增量同步，你可以訪問這個頁面信息。

另外，我想提一下 Meltano Hub 真的很棒。它易于導(dǎo)航并找到你需要的內(nèi)容。此外，你可以通過查看有多少客戶使用它們來輕松比較不同的加載器或抽取器。

如果你對構(gòu)建基于 AI 的應(yīng)用程序感興趣，請在以下博客文章中查找更多信息：

• 在你的數(shù)據(jù)集上實(shí)現(xiàn)全文和語義搜索能力。
• 連接你的數(shù)據(jù)與 LLMs，構(gòu)建問題 - 答案。
• 構(gòu)建一個使用檢索增強(qiáng)生成（RAG）模式的聊天機(jī)器人。

準(zhǔn)備將 RAG 構(gòu)建到你的應(yīng)用中了嗎？想要嘗試不同的 LLMs 與向量數(shù)據(jù)庫嗎？ 查看我們在 Github 上關(guān)于 LangChain、Cohere 等的示例 notebooks，并加入即將開始的 Elasticsearch 工程師培訓(xùn)！

原文：Ingest Data from Snowflake to Elasticsearch using Meltano: A developer’s journey — Elastic Search Labs