Передача данных из эндпоинта-источника Yandex Data Streams

С помощью сервиса Yandex Data Transfer вы можете переносить данные из очереди Data Streams и реализовывать различные сценарии переноса, обработки и трансформации данных. Для реализации трансфера:

1. Ознакомьтесь с возможными сценариями передачи данных.
2. Подготовьте базу данных Data Streams к трансферу.
3. Настройте эндпоинт-источник в Yandex Data Transfer.
4. Настройте один из поддерживаемых приемников данных.
5. Cоздайте и запустите трансфер.
6. Выполняйте необходимые действия по работе с базой и контролируйте трансфер.
7. При возникновении проблем, воспользуйтесь готовыми решениями по их устранению.

Сценарии передачи данных из Data StreamsСценарии передачи данных из Data Streams

1. Миграция — перенос данных из одного хранилища в другое. Часто это перенос базы из устаревших локальных баз в управляемые облачные.

   Отдельной задачей миграции является зеркалирование данных между очередями Data Streams.

2. Поставка данных — процесс доставки произвольных данных в целевые хранилища. Процесс поставки включает извлечение данных из очереди и их десериализацию с последующей трансформацией данных в формат целевого хранилища.

   • YDS в ClickHouse®;
   • YDS в Object Storage;
   • YDS в OpenSearch.

Подробное описание возможных сценариев передачи данных в Yandex Data Transfer см. в разделе Практические руководства.

Подготовка базы данных источникаПодготовка базы данных источника

1. Создайте сервисный аккаунт с ролью yds.editor.

2. Создайте поток данных.

3. (Опционально) Создайте функцию обработки.

   Пример функции обработки

   const yc = require("yandex-cloud");
   const { Parser } = require("@robojones/nginx-log-parser");
   module.exports.handler = async function (event, context) {
       const schema =
           '$remote_addr - $remote_user [$time_local] "$request" $status $bytes_sent "$http_referer" "$http_user_agent"';
       const parser = new Parser(schema);
       return {
           Records: event.Records.map((record) => {
               const decodedData = new Buffer(record.kinesis.data, "base64")
                   .toString("ascii")
                   .trim();
               try {
                   const result = parser.parseLine(decodedData);
                   if (result.request == "") {
                       // empty request - drop message
                       return {
                           eventID: record.eventID,
                           invokeIdentityArn: record.invokeIdentityArn,
                           eventVersion: record.eventVersion,
                           eventName: record.eventName,
                           eventSourceARN: record.eventSourceARN,
                           result: "Dropped"
                       };
                   }
                   return {
                       // successfully parsed message
                       eventID: record.eventID,
                       invokeIdentityArn: record.invokeIdentityArn,
                       eventVersion: record.eventVersion,
                       eventName: record.eventName,
                       eventSourceARN: record.eventSourceARN,
                       kinesis: {
                           data: new Buffer(JSON.stringify(result)).toString(
                               "base64"
                           ),
                       },
                       result: "Ok"
                   };
               } catch (err) {
                   // error - fail message
                   return {
                       eventID: record.eventID,
                       invokeIdentityArn: record.invokeIdentityArn,
                       eventVersion: record.eventVersion,
                       eventName: record.eventName,
                       eventSourceARN: record.eventSourceARN,
                       result: "ProcessingFailed",
                   };
               }
           })
       };
   };
   

4. (Опционально) Подготовьте файл схемы данных в формате JSON.

   Пример файла со схемой данных:

   [
       {
           "name": "<имя_поля>",
           "type": "<тип>"
       },
       ...
       {
           "name": "<имя_поля>",
           "type": "<тип>"
       }
   ]
   

   Список допустимых типов:

• any
• boolean
• datetime
• double
• int8
• int16
• int32
• int64
• string
• uint8
• uint16
• uint32
• uint64
• utf8

Настройка эндпоинта-источника Data StreamsНастройка эндпоинта-источника Data Streams

При создании или изменении эндпоинта вы можете задать:

• Настройки подключения к потоку данных в Yandex Data Streams. Это обязательные параметры.
• Расширенные настройки.

Основные настройкиОсновные настройки

Расширенные настройкиРасширенные настройки

В расширенных настройках вы можете задать правила трансформации и правила конвертации. Данные обрабатываются в следующем порядке:

1. Трансформация. Данные в формате JSON передаются функции Yandex Cloud Functions. В теле функции содержатся метаинформация и необработанные данные, которые переданы в очередь. С помощью функции данные обрабатываются и возвращаются в Data Transfer.

2. Конвертация. Выполняется парсинг, с помощью которого данные подготавливаются для передачи приемнику.

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

Настройка приемника данныхНастройка приемника данных

Настройте один из поддерживаемых приемников данных:

• PostgreSQL;
• MySQL®;
• MongoDB
• ClickHouse®;
• Greenplum®;
• Yandex Managed Service for YDB;
• Yandex Object Storage;
• Apache Kafka®;
• YDS;
• Elasticsearch;
• OpenSearch.

Полный список поддерживаемых источников и приемников в Yandex Data Transfer см. в разделе Доступные трансферы.

После настройки источника и приемника данных создайте и запустите трансфер.

Решение проблем, возникающих при переносе данныхРешение проблем, возникающих при переносе данных

• Прерывание трансфера с ошибкой
• Редиректы Cloud Functions

См. полный список рекомендаций в разделе Решение проблем.

Прерывание трансфера с ошибкойПрерывание трансфера с ошибкой

Трансфер типа Репликация или Копирование и репликация прерывается с ошибкой.

Текст ошибки:

/Ydb.PersQueue.V1.PersQueueService/AddReadRule failed: OVERLOADED

Трансфер прерывается из-за ограничения облачной квоты на количество операций с Managed Service for YDB.

Решение:

1. Увеличьте в квотах Managed Service for YDB на облако с нужной базой данных значение характеристики Количество схемных операций в минуту и активируйте трансфер повторно.

Редиректы Cloud FunctionsРедиректы Cloud Functions

В трансферах из Data Streams или Apache Kafka® в редких случаях может возникнуть ошибка:

redirect to SOME_URL is requested but no redirects are allowed.

Возможная причина:

На источнике настроено использование функции Cloud Functions, которая возвращает не данные, а редирект на другой URL.

Решение:

По соображениям безопасности такие редиректы запрещены. Воздержитесь от использования редиректов в Cloud Functions в трансферах.