Создание коннектора

Веб-консоль
MWS CLI
API

В веб-консоли выберите нужный проект.
В списке сервисов выберите Managed Kafka.
Нажмите на имя нужного кластера и перейдите на вкладку Коннекторы.
Нажмите кнопку Создать.
(опционально) Измените автоматически сгенерированное имя коннектора. Позже это имя (name) и созданный из него идентификатор (ID) нельзя изменить, меняется только отображаемое имя (display name) коннектора.
Выберите, как будут выбираться топики для чтения данных: Конкретные топики или Регулярное выражение.
Укажите конкретные топики или шаблон, которому должны соответствовать имена топиков.

Если в кластере еще нет нужного топика, создайте его.
(опционально) Добавьте описание для коннектора.
Нажмите кнопку Далее.
Настройте доступ к бакету:
1. Оставьте эндпоинт MWS Object Storage или укажите эндпоинт стороннего S3-совместимого хранилища.
2. Укажите бакет, в который будут экспортироваться данные.
  
  Если нужного бакета в MWS Object Storage еще нет, создайте его.
3. Добавьте идентификатор (Access Key) и секретную часть (Secret Key) существующего HMAC-ключа.
  
  Если для сервисного аккаунта в MWS Cloud Platform еще нет HMAC-ключа, создайте его.
4. (опционально) Настройте доступ к бакету.
5. Нажмите кнопку Далее.
Задайте параметры обработки данных:
1. Укажите максимальное количество параллельно выполняемых задач. Уменьшить значение в будущем невозможно. Значение не должно превышать общее количество разделов во всех выбранных топиках.
2. Выберите формат входных данных для десериализации сообщений из топика. Изменить формат после создания коннектора будет невозможно.
3. Выберите тип сжатия для файлов, которые будут сохранены в бакет.
4. (опционально) Включите опцию Оборачивать значения в JSON-объект.
5. Укажите поля сообщения из топиков, которые будут записаны в сохраняемый в бакет файл.
6. Укажите DLQ-топик для отправки сообщений, обработанных с ошибками.
  
  Если нужного топика еще нет, создайте его.
7. Нажмите кнопку Далее.
Проверьте заданные настройки и нажмите кнопку Создать.

Если у вас нет утилиты MWS CLI:
1. Установите утилиту.
2. Инициализируйте профиль.
Узнайте имя кластера:
bash
```
1
mws mkafka cluster list
```
Узнайте имена DLQ-топика и топиков для записи:
bash
```
1
mws mkafka topic list --cluster <имя кластера>
```

Выполните команду:

1
mws mkafka connector create <имя коннектора> \
2
  --cluster <имя кластера> \
3
  --display-name <отображаемое имя коннектора> \
4
  --s3-sink-connector-topics <имя топика 1>,<имя топика 2> \  # Если указываются конкретные топики
5
  --s3-sink-connector-topics-regex ^<префикс-топика>-.*$ \  # Если указывается регулярное выражение
6
  --s3-sink-connector-s3-properties-bucket-name <имя бакета для записи данных> \
7
  --s3-sink-connector-s3-properties-access-key-id <идентификатор (Acess Key) HMAC-ключа> \
8
  --s3-sink-connector-s3-properties-secret-access-key <секретная часть (Secret Key) HMAC-ключа> \
9
  --s3-sink-connector-s3-properties-endpoint <S3-эндпоинт> \
10
  --s3-sink-connector-tasks-max <максимальное количество параллельно выполняемых задач> \  # Значение не должно превышать общее количество разделов во всех выбранных топиках
11
  --s3-sink-connector-input-format <формат входных данных — STRING или JSON> \  # Изменить значение после создания коннектора будет невозможно
12
  --s3-sink-connector-output-format <формат выходных данных — JSONL> \
13
  --s3-sink-connector-file-compression-type <тип сжатия файлов — GZIP, SNAPPY, ZSTD или NONE> \
14
  --s3-sink-connector-output-envelope \  # Если нужно сохранять данные как JSON-объект
15
  --s3-sink-connector-output-fields <поле для записи — KEY, VALUE, OFFSET, TIMESTAMP или HEADERS> \  # Если данные не сохраняются как JSON-объект, должно содержать только одно поле. Если данные сохраняются как JSON-объект — параметр опционален. Чтобы указать несколько полей, повторите флаг несколько раз
16
  --s3-sink-connector-error-handling <политика обработки ошибок — DLQ> \
17
  --s3-sink-connector-dlq-topic <имя DLQ-топика> \
18
  --active

1
metadata:
2
    createTime: "2025-10-22T04:04:51Z"
3
    updateTime: "2025-10-22T04:04:51Z"
4
    id: mkafka/projects/<имя проекта>/clusters/<имя кластера>/topics/<имя топика>
5
spec:
6
    partitions: 2
7
    replicationFactor: 1
8
status:
9
    ready:
10
        state: PROCESSING

Убедитесь, что у вас есть IAM-токен для авторизации запроса. Если нет, получите его.

Узнайте имя кластера:

1
curl -X GET https://mkafka.mwsapis.ru/mkafka/v1/projects/<имя проекта>/clusters \
2
  -H "Authorization: Bearer <IAM-токен>"

Узнайте имена DLQ-топика и топиков для записи:

1
curl -X GET https://mkafka.mwsapis.ru/mkafka/v1/projects/<имя проекта>/clusters/<имя кластера>/topics \
2
  -H "Authorization: Bearer <IAM-токен>"

Опишите характеристики коннектора в виде JSON-объекта:

1
{
2
  "spec": {
3
    "s3SinkConnector": {
4
      "displayName": "<отображаемое имя коннектора>",
5
      "topics": "<имя топика 1>,<имя топика 2>", // Если указываются конкретные топики
6
      "topicsRegex": "^<префикс-топика>-.*$", // Если указывается регулярное выражение
7
      "s3Properties": {
8
        "bucketName": "<имя бакета для записи данных>",
9
        "accessKeyId": "<идентификатор (Acess Key) HMAC-ключа>",
10
        "secretAccessKey": "<секретная часть (Secret Key) HMAC-ключа>",
11
        "endpoint": "<S3-эндпоинт>"
12
      },
13
      "tasksMax": <максимальное количество параллельно выполняемых задач>, // Значение не должно превышать общее количество разделов во всех выбранных топиках
14
      "inputFormat": "<формат входных данных — STRING или JSON>", // Изменить значение после создания коннектора будет невозможно
15
      "outputFormat": "<формат выходных данных — JSONL>",
16
      "fileCompressionType": "<тип сжатия файлов — GZIP, SNAPPY, ZSTD или NONE>",
17
      "outputEnvelope": <сохранять данные как JSON-объект — true или false>,
18
      "outputFields": ["<поле для записи 1 — KEY, VALUE, OFFSET, TIMESTAMP или HEADERS>","<поле для записи 2 — KEY, VALUE, OFFSET, TIMESTAMP или HEADERS>"], // Если данные сохраняются как JSON-объект — параметр опционален. Если данные не сохраняются как JSON-объект, должно содержать только одно поле
19
      "errorHandling": "<политика обработки ошибок — DLQ>",
20
      "dlqTopic": "<имя DLQ-топика>"
21
    }
22
  }
23
}

Выполните запрос:

1
curl -X POST https://mkafka.mwsapis.ru/mkafka/v1/projects/<имя проекта>/clusters/<имя кластера>/connector/<имя коннектора> \
2
  -H "Authorization: Bearer <IAM-токен>" \
3
  -H "Content-Type: application/json" \
4
  -d '@<путь к файлу с JSON-объектом>'

Если запрос был успешно принят, вы получите ответ с кодом 200 OK. Если коннектор создан успешно, в ответе будут выведены параметры нового коннектора.