连接器 API

了解如何使用Elasticsearch Connector APIs设置自管理连接器。

对于这个示例，我们将使用 connectors-postgresql，PostgreSQL 连接器将数据从 PostgreSQL 数据库同步到 Elasticsearch。 我们将在 Docker 中启动一个带有示例数据的简单 PostgreSQL 实例，创建一个连接器，并将数据同步到 Elasticsearch。 您可以按照相同的步骤为另一个数据源设置连接器。

如果您已经在 Elastic Cloud 上部署了 Elasticsearch（托管部署或无服务器项目），您就可以开始了。 要在 Docker 中以本地开发模式启动 Elasticsearch 进行测试，请打开下面的可折叠部分。

docker run -p 9200:9200 -d --name elasticsearch \
  -e "discovery.type=single-node" \
  -e "xpack.security.enabled=false" \
  -e "xpack.security.http.ssl.enabled=false" \
  -e "xpack.license.self_generated.type=trial" \
  docker.elastic.co/elasticsearch/elasticsearch:9.0.0-beta1

export ELASTIC_PASSWORD="changeme"

curl -s -X GET -u elastic:$ELASTIC_PASSWORD http://localhost:9200

curl -s -X PUT http://localhost:9200/_connector/my-connector-id \
-H "Authorization: APIKey $APIKEY" \
-H "Content-Type: application/json" \
-d '{
  "name": "Music catalog",
  "index_name":  "music",
  "service_type": "postgresql"
}'

在本教程中，我们将在Docker中设置一个带有示例数据的PostgreSQL实例。 当然，如果您已经有一个PostgreSQL实例，可以跳过此步骤并使用您自己的现有PostgreSQL实例。 请记住，使用不同的实例可能需要在下一步中对连接器配置进行调整。

docker run --name postgres -e POSTGRES_USER=myuser -e POSTGRES_PASSWORD=mypassword -p 5432:5432 -d postgres

mkdir -p ~/data

curl -L https://raw.githubusercontent.com/lerocha/chinook-database/master/ChinookDatabase/DataSources/Chinook_PostgreSql.sql -o ~/data/Chinook_PostgreSql.sql

docker cp ~/data/Chinook_PostgreSql.sql postgres:/
docker exec -it postgres psql -U myuser -f /Chinook_PostgreSql.sql

docker exec -it postgres psql -U myuser -d chinook -c "\dt"

本教程使用了一个非常基础的设置。要使用诸如过滤规则和增量同步等高级功能，请在您的PostgreSQL数据库上启用track_commit_timestamp。更多详情请参阅postgresql-connector-client-tutorial。

现在是真正有趣的时候了！我们将设置一个连接器，在Elasticsearch中创建一个可搜索的PostgreSQL数据镜像。

我们将使用创建连接器API来创建一个PostgreSQL连接器实例。

使用开发工具控制台或curl运行以下API调用：

PUT _connector/my-connector-id
{
  "name": "Music catalog",
  "index_name":  "music",
  "service_type": "postgresql"
}

请注意，我们在PUT请求中指定了my-connector-id ID。 我们需要连接器ID来在本地设置和运行连接器服务。

如果您更喜欢使用自动生成的ID，请将 PUT _connector/my-connector-id 替换为 POST _connector。

现在我们将运行连接器服务，以便我们可以开始将数据从PostgreSQL实例同步到Elasticsearch。 我们将使用在connectors-run-from-docker中概述的步骤。

当在您自己的基础设施上运行连接器服务时，您需要提供一个包含以下详细信息的配置文件：

如果您还没有创建用于访问 Elasticsearch 的 API 密钥，可以使用 _security/api_key 端点。

在这里，我们假设您的目标 Elasticsearch 索引名称是 music。如果您使用不同的索引名称，请相应地调整请求体。

POST /_security/api_key
{
  "name": "music-connector",
  "role_descriptors": {
    "music-connector-role": {
      "cluster": [
        "monitor",
        "manage_connector"
      ],
      "indices": [
        {
          "names": [
            "music",
            ".search-acl-filter-music",
            ".elastic-connectors*"
          ],
          "privileges": [
            "all"
          ],
          "allow_restricted_indices": false
        }
      ]
    }
  }
}

您需要使用响应中的 encoded 值作为配置文件中的 elasticsearch.api_key。

让我们创建一个目录和一个config.yml文件来存储连接器配置：

mkdir -p ~/connectors-config
touch ~/connectors-config/config.yml

现在，让我们将连接器详细信息添加到配置文件中。 打开 config.yml 并粘贴以下配置，将占位符替换为您自己的值：

elasticsearch.host: <ELASTICSEARCH_ENDPOINT> # Your Elasticsearch endpoint
elasticsearch.api_key: <ELASTICSEARCH_API_KEY> # Your Elasticsearch API key

connectors:
  - connector_id: "my-connector-id"
    service_type: "postgresql"

我们提供了一个示例配置文件，供参考，位于elastic/connectors仓库中。

现在我们已经配置好了配置文件，我们可以在本地运行连接器服务。 这将使您的连接器实例指向您的 Elasticsearch 部署。

运行以下 Docker 命令以启动连接器服务：

docker run \
-v "$HOME/connectors-config:/config" \
--rm \
--tty -i \
--network host \
docker.elastic.co/integrations/elastic-connectors:9.0.0-beta1.0 \
/app/bin/elastic-ingest \
-c /config/config.yml

通过获取连接器状态（应为needs_configuration）和last_seen字段来验证您的连接器是否已连接（注意时间以UTC报告）。 last_seen字段表示连接器已成功连接到Elasticsearch。

GET _connector/my-connector-id

现在我们的连接器实例已经启动并运行，但它还不知道从哪里同步数据。 最后一步是配置我们的连接器，提供关于我们的PostgreSQL实例的详细信息。 在Elastic Cloud或Serverless UI中设置连接器时，系统会提示您在用户界面中添加这些详细信息。

但因为本教程是关于以编程方式使用连接器，我们将使用更新连接器配置API来添加我们的配置细节。

运行以下 API 调用以使用我们的 connectors-postgresql-client-configuration 配置连接器，PostgreSQL 配置详情：

PUT _connector/my-connector-id/_configuration
{
  "values": {
    "host": "127.0.0.1",
    "port": 5432,
    "username": "myuser",
    "password": "mypassword",
    "database": "chinook",
    "schema": "public",
    "tables": "album,artist"
  }
}

我们现在准备将PostgreSQL数据同步到Elasticsearch。 运行以下API调用来启动完全同步作业：

POST _connector/_sync_job
{
    "id": "my-connector-id",
    "job_type": "full"
}

要在 Elasticsearch 中存储数据，连接器需要创建一个索引。 当我们创建连接器时，我们指定了 music 索引。 连接器将在启动同步作业之前创建并配置此 Elasticsearch 索引。

使用获取同步作业 API来跟踪同步作业的状态和进度。 默认情况下，首先返回最新的作业状态。 运行以下 API 调用来检查同步作业的状态：

GET _connector/_sync_job?connector_id=my-connector-id&size=1

随着同步的进行，作业文档将会被更新，您可以随时检查以获取更新。

一旦作业完成，状态应为completed，并且indexed_document_count应为622。

验证数据是否存在于 music 索引中，使用以下 API 调用：

GET music/_count

Elasticsearch 将数据存储在文档中，这些文档是 JSON 对象。 使用以下 API 调用列出各个文档：

GET music/_search

使用以下命令检查最新同步作业的状态：

GET _connector/_sync_job?connector_id=my-connector-id&size=1

如果在同步过程中连接器遇到任何错误，您将在 error 字段中找到这些错误。

要删除连接器及其关联的同步作业，请运行以下命令：

DELETE _connector/my-connector-id&delete_sync_jobs=true

这不会删除连接器创建的用于存储数据的Elasticsearch索引。 通过运行以下命令删除music索引：

DELETE music

要删除 PostgreSQL 容器，请运行以下命令：

docker stop postgres
docker rm postgres

要移除连接器服务，请运行以下命令：

docker stop <container_id>
docker rm <container_id>

恭喜！您已成功使用Connector API设置了一个自管理的连接器。

以下是一些可供探索的下一步：