Практическое руководство по API соединителей

Узнайте, как настроить управляемый пользователем коннектор, используя API коннекторов Elasticsearch.

В этом примере мы будем использовать коннектор connectors-postgresql,PostgreSQL для синхронизации данных из базы данных PostgreSQL с Elasticsearch. Мы запустим простой экземпляр PostgreSQL в Docker с некоторыми тестовыми данными, создадим коннектор и синхронизируем данные с Elasticsearch. Вы можете выполнить те же шаги для настройки коннектора для другого источника данных.

Если у вас уже есть развертывание Elasticsearch в Elastic Cloud (Развертывание хостинга или Сервисный проект), вы можете начать работу. Чтобы запустить Elasticsearch в режиме разработки локально в Docker для целей тестирования, откройте раскрывающийся раздел ниже.

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:8.17.3

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"
}'

Для этого учебника мы настроим экземпляр PostgreSQL в Docker с некоторыми тестовыми данными. Конечно, вы можете пропустить этот шаг и использовать свой собственный существующий экземпляр 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"

Этот учебник использует очень простую настройку. Для использования расширенных функций, таких как правила фильтрации и инкрементная синхронизация, включите track_commit_timestamp в вашей базе данных PostgreSQL. Обратитесь к postgresql-connector-client-tutorial для получения дополнительных сведений.

Теперь пришло время для настоящего интереса! Мы настроим коннектор для создания индексируемого зеркала наших данных PostgreSQL в Elasticsearch.

Мы будем использовать API создания коннектора для создания экземпляра коннектора PostgreSQL.

Выполните следующий API-вызов, используя Консоль инструментов разработчика или curl:

resp = client.connector.put(
    connector_id="my-connector-id",
    name="Music catalog",
    index_name="music",
    service_type="postgresql",
)
print(resp)

const response = await client.connector.put({
  connector_id: "my-connector-id",
  name: "Music catalog",
  index_name: "music",
  service_type: "postgresql",
});
console.log(response);

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

Обратите внимание, что мы указали идентификатор my-connector-id как часть запроса PUT. Нам понадобится идентификатор коннектора для настройки и запуска службы коннектора локально.

Если вы предпочитаете использовать автоматически сгенерированный идентификатор, замените PUT _connector/my-connector-id на POST _connector.

Теперь мы запустим службу коннектора, чтобы начать синхронизацию данных из нашего экземпляра PostgreSQL с Elasticsearch. Мы будем использовать шаги, описанные в connectors-run-from-docker.

При запуске службы коннектора на собственной инфраструктуре вам необходимо предоставить файл конфигурации со следующими данными:

Если вы еще не создали ключ API для доступа к Elasticsearch, вы можете использовать конечную точку _security/api_key.

Здесь мы предполагаем, что имя целевого индекса Elasticsearch — music. Если вы используете другое имя индекса, скорректируйте тело запроса соответственно.

resp = client.security.create_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
                }
            ]
        }
    },
)
print(resp)

const response = await client.security.createApiKey({
  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,
        },
      ],
    },
  },
});
console.log(response);

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/enterprise-search/elastic-connectors:8.17.3.0 \
/app/bin/elastic-ingest \
-c /config/config.yml

Проверьте подключение коннектора, получив статус коннектора (должен быть needs_configuration) и поле last_seen (обратите внимание, что время указано в UTC). Поле last_seen указывает на успешное подключение коннектора к Elasticsearch.

resp = client.connector.get(
    connector_id="my-connector-id",
)
print(resp)

const response = await client.connector.get({
  connector_id: "my-connector-id",
});
console.log(response);

GET _connector/my-connector-id

Теперь наш экземпляр коннектора запущен, но он ещё не знает, откуда синхронизировать данные. Последним шагом является настройка коннектора с деталями нашего экземпляра PostgreSQL. При настройке коннектора в интерфейсах Elastic Cloud или Serverless вы будете перенаправлены на добавление этих деталей в пользовательском интерфейсе.

Но так как этот учебник посвящён работе с коннекторами программно, мы воспользуемся API для обновления конфигурации коннектора, чтобы добавить необходимые параметры конфигурации.

Выполните следующий вызов API для настройки коннектора с параметрами конфигурации connectors-postgresql-client-configuration, PostgreSQL:

resp = client.connector.update_configuration(
    connector_id="my-connector-id",
    values={
        "host": "127.0.0.1",
        "port": 5432,
        "username": "myuser",
        "password": "mypassword",
        "database": "chinook",
        "schema": "public",
        "tables": "album,artist"
    },
)
print(resp)

const response = await client.connector.updateConfiguration({
  connector_id: "my-connector-id",
  values: {
    host: "127.0.0.1",
    port: 5432,
    username: "myuser",
    password: "mypassword",
    database: "chinook",
    schema: "public",
    tables: "album,artist",
  },
});
console.log(response);

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 для запуска полного задания синхронизации:

resp = client.perform_request(
    "POST",
    "/_connector/_sync_job",
    headers={"Content-Type": "application/json"},
    body={
        "id": "my-connector-id",
        "job_type": "full"
    },
)
print(resp)

const response = await client.transport.request({
  method: "POST",
  path: "/_connector/_sync_job",
  body: {
    id: "my-connector-id",
    job_type: "full",
  },
});
console.log(response);

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

Для хранения данных в Elasticsearch коннектору необходимо создать индекс. При создании коннектора мы указали индекс music. Коннектор создаст и настроит этот индекс Elasticsearch перед запуском задания синхронизации.

Используйте API для получения задания синхронизации для отслеживания статуса и прогресса задания синхронизации. По умолчанию самые последние статусы заданий возвращаются первыми. Выполните следующий вызов API, чтобы проверить статус задания синхронизации:

resp = client.perform_request(
    "GET",
    "/_connector/_sync_job",
    params={
        "connector_id": "my-connector-id",
        "size": "1"
    },
)
print(resp)

const response = await client.transport.request({
  method: "GET",
  path: "/_connector/_sync_job",
  querystring: {
    connector_id: "my-connector-id",
    size: "1",
  },
});
console.log(response);

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

Документ задания будет обновляться по мере прогресса синхронизации, вы можете проверять его так часто, как хотите.

После завершения задания статус должен быть completed, а indexed_document_count должен быть 622.

Проверьте наличие данных в индексе music с помощью следующего вызова API:

resp = client.count(
    index="music",
)
print(resp)

const response = await client.count({
  index: "music",
});
console.log(response);

GET music/_count

Elasticsearch хранит данные в документах, которые являются JSON-объектами. Перечислите отдельные документы с помощью следующего вызова API:

resp = client.search(
    index="music",
)
print(resp)

const response = await client.search({
  index: "music",
});
console.log(response);

GET music/_search

Используйте следующую команду для проверки статуса последнего задания синхронизации:

resp = client.perform_request(
    "GET",
    "/_connector/_sync_job",
    params={
        "connector_id": "my-connector-id",
        "size": "1"
    },
)
print(resp)

const response = await client.transport.request({
  method: "GET",
  path: "/_connector/_sync_job",
  querystring: {
    connector_id: "my-connector-id",
    size: "1",
  },
});
console.log(response);

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

Если во время синхронизации коннектор столкнулся с ошибками, вы найдёте их в поле error.

Чтобы удалить коннектор и связанные с ним задания синхронизации, выполните эту команду:

resp = client.connector.delete(
    connector_id="my-connector-id&delete_sync_jobs=true",
)
print(resp)

const response = await client.connector.delete({
  connector_id: "my-connector-id&delete_sync_jobs=true",
});
console.log(response);

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

Это не удалит индекс Elasticsearch, созданный коннектором для хранения данных. Удалите индекс music, выполнив следующую команду:

resp = client.indices.delete(
    index="music",
)
print(resp)

const response = await client.indices.delete({
  index: "music",
});
console.log(response);

DELETE music

Чтобы удалить контейнер PostgreSQL, выполните следующие команды:

docker stop postgres
docker rm postgres

Чтобы удалить службу коннектора, выполните следующие команды:

docker stop <container_id>
docker rm <container_id>

Поздравляем! Вы успешно настроили самостоятельный коннектор с помощью API коннекторов.

Вот некоторые следующие шаги для изучения: