Elasticsearch：将数据从 Snowflake 摄取到 Elasticsearch

作者：来自 Elastic Ashish Tiwari

为了利用 Elasticsearch® 提供的强大搜索功能，许多企业在 Elasticsearch 中保留可搜索数据的副本。 Elasticsearch 是一种经过验证的技术，适用于传统文本搜索以及用于语义搜索用例的向量搜索。 Elasticsearch Relevance EngineTM (ESRE) 使你能够在专有数据上添加语义搜索，这些数据可以与生成式 AI 技术集成以构建现代搜索体验。

Snowflake 是一种完全托管的 SaaS（软件即服务），为数据仓库、数据湖、数据工程、数据科学、数据应用程序开发以及实时/共享数据的安全共享和消费提供单一平台。

在本博客中，我们将了解如何使用以下方法将 snowflake 数据导入 Elasticsearch：

1. 使用 Logstash®（定期同步）
2. 使用 Snowflake Elasticsearch Python 脚本（一次性同步）

先决条件

Snowflake 凭证

注册后你将收到以下所有凭据，或者你可以从 Snowflake 面板获取它们。

• 账户用户名
• 账号密码
• 账户标识符

Elastic® 凭证

1. 访问 https://cloud.elastic.co 并注册。
2. 单击 Create deployment。 在弹出窗口中，你可以更改设置或保留默认设置。
3. 下载或复制部署凭据（用户名和密码）。
4. 同时复制 Cloud ID。
5. 准备好部署后，单击 “Continue”（或单击 “Open Kibana”）。 它会将你重定向到 Kibana® 仪表板。

使用 Logstash

Logstash 是一个免费且开放的 ETL 工具，你可以在其中提供多个源作为输入、转换（修改）它并推送到你最喜欢的存储。 Logstash 的著名用例之一是从文件中读取日志并将其推送到 Elasticsearch。 我们还可以使用 filter 插件动态修改数据，它会将更新的数据推送到输出。

我们将使用 JDBC input plugin 从 Snowflake 中提取数据，并使用 Elasticsearch output plugin 将数据推送到 Elasticsearch。

1. 参考文档安装 Logstash。
2. 转到 Maven 中央存储库并下载：https://repo1.maven.org/maven2/net/snowflake/snowflake-jdbc。
3. 单击你需要的版本的目录并下载 snowflake-jdbc-#.#.#.jar 文件。 就我而言，我已经下载了snowflake-jdbc-3.9.2.jar。 （有关 Snowflake JDBC 驱动程序的更多信息，请参阅官方文档。）
4. 通过创建文件 sf-es.conf 创建管道。 添加以下代码片段并替换所有凭证。

sf-es.conf

input {
  jdbc {
    jdbc_driver_library => "/usr/share/logstash/logstash_external_configs/driver/snowflake-jdbc-3.9.2.jar"
    jdbc_driver_class => "net.snowflake.client.jdbc.SnowflakeDriver"
    jdbc_connection_string => "jdbc:snowflake://<account_identifier>.snowflakecomputing.com/?db=SNOWFLAKE_SAMPLE_DATA&warehouse=COMPUTE_WH&schema=TPCH_SF1"
    jdbc_user => "<snowflake_username>"
    jdbc_password => "<snowflake_password>"
    schedule => "* * * * *"
    statement => "select * from customer limit 10;"
  }
}

filter {}

output {
  elasticsearch {
    cloud_id => "<elastic cloud_id>"
    cloud_auth => "<elastic_username>:<elastic_password>"
    index => "sf_customer"
  }
}

jdbc_connection_string ：

db=SNOWFLAKE_SAMPLE_DATA
warehouse=COMPUTE_WH
schema=TPCH_SF1

Schedule：你可以在此处计划使用 cron 语法定期运行此流程。 每次运行时，你的数据都会增量移动。 你可以查看有关 scheduling 的更多信息。

请根据你的要求进行更改。

JDBC 分页（可选）：这将导致一条 sql 语句被分解为多个查询。 每个查询将使用限制和偏移量来共同检索完整的结果集。 你可以使用它在一次运行中移动所有数据。

通过添加以下配置来启用 JDBC 分页：

jdbc_paging_enabled => true,
jdbc_paging_mode => "explicit",
jdbc_page_size => 100000

        5. 运行 Logstash

bin/logstash -f sf-es.conf

Snowflake-Elasticsearch Python 脚本

如果 Logstash 当前尚未到位或尚未实现，我编写了一个小型 Python 实用程序（可在 GitHub 上找到）从 Snowflake 提取数据并将其推送到 Elasticsearch。 这将一次性提取你的所有数据。 因此，如果你有少量数据需要非定期迁移，则可以使用此实用程序。

1） 安装

git clone https://github.com/ashishtiwari1993/snowflake-elasticsearch-connector.git
cd snowflake-elasticsearch-connector

2）安装依赖项

git clone https://github.com/ashishtiwari1993/snowflake-elasticsearch-connector.git
cd snowflake-elasticsearch-connector

3）修改配置

• 打开 config/connector.yml。
• 将凭证替换为以下内容：

snowflake:
  username: <sf_username>
  password: <sf_password>
  account: <sf_account_identifier>
  database: <db_name>
  table: <table_name>
  columns: ""
  warehouse: ""
  scheme: ""
  limit: 50

elasticsearch:
  host: https://localhost:9200
  username: elastic
  password: elastic@123
  ca_cert: /path/to/elasticsearch/config/certs/http_ca.crt
  index: <sf_customer>

4）运行 Connector

python __main__.py

验证数据

登录 Kibana 并转到 ☰ > Management > Dev Tools。

将以下 API GET 请求复制并粘贴到控制台窗格中，然后单击 ▶（播放）按钮。 这会查询新索引中的所有记录。

GET sf_customer/_search
{
  "query": {
    "match_all": {}
  }
}

结论

我们已成功将数据从 Snowflake 迁移到 Elastic Cloud。 你可以在任何 Elasticsearch 实例上实现相同的目标，无论是在云端还是在本地。

开始在你的数据集上利用全文和语义搜索功能。 你还可以将你的数据与 LLM 连接起来，以构建问答功能。