Миграция данных из Yandex Data Streams с помощью Yandex Data Transfer

С помощью сервиса Data Transfer вы можете перенести данные из потока Data Streams в бакет Yandex Object Storage:

1. Подготовьте поток данных Data Streams.
2. Подготовьте и активируйте трансфер.
3. Проверьте работоспособность трансфера.

Если созданные ресурсы вам больше не нужны, удалите их.

Перед началом работыПеред началом работы

Подготовьте инфраструктуру:

Подготовьте поток данных Data StreamsПодготовьте поток данных Data Streams

1. Создайте поток данных Data Streams.

2. Отправьте в поток тестовые данные. В качестве сообщения используйте данные от сенсоров автомобиля в формате JSON:

{
    "device_id":"iv9a94th6rzt********",
    "datetime":"2020-06-05T17:27:00",
    "latitude":"55.70329032",
    "longitude":"37.65472196",
    "altitude":"427.5",
    "speed":"0",
    "battery_voltage":"23.5",
    "cabin_temperature":"17",
    "fuel_level":null
}

Подготовьте и активируйте трансферПодготовьте и активируйте трансфер

1. Создайте эндпоинт для источника:

   • Тип базы данных — Yandex Data Streams.

   • Параметры эндпоинта:

     • Настройки подключения:

       • База данных — выберите базу данных Managed Service for YDB из списка.
       • Поток — укажите имя потока Data Streams.
       • Сервисный аккаунт — выберите или создайте сервисный аккаунт с ролью yds.editor.

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

       • Правила конвертации — JSON.

       • Схема данных — JSON-спецификация:

         Создайте и загрузите файл схемы данных в формате JSON json_schema.json:

         json_schema.json

         [
             {
                 "name": "device_id",
                 "type": "string"
             },
             {
                 "name": "datetime",
                 "type": "datetime"
             },
             {
                 "name": "latitude",
                 "type": "double"
             },
             {
                 "name": "longitude",
                 "type": "double"
             },
             {
                 "name": "altitude",
                 "type": "double"
             },
             {
                 "name": "speed",
                 "type": "double"
             },
             {
                 "name": "battery_voltage",
                 "type": "any"
             },
             {
                 "name": "cabin_temperature",
                 "type": "double"
             },
             {
                 "name": "fuel_level",
                 "type": "any"
             }
         ]
         

2. Создайте эндпоинт для приемника:

   • Тип базы данных — Object Storage.

   • Параметры эндпоинта:

     • Настройки подключения:

       • Бакет — укажите имя бакета в Object Storage.
       • Сервисный аккаунт — выберите или создайте сервисный аккаунт с ролью storage.uploader.

     • Выходной формат — выберите JSON или CSV, если в расширенных настройках эндпоинта источника включили опцию Правила конвертации.

3. Создайте трансфер:

   Вручную

   Terraform

   1. Создайте трансфер типа Репликация, использующий созданные эндпоинты.
   2. Активируйте его.

   1. Раскомментируйте в файле data-transfer-yds-obj.tf:

      • переменные:

        • source_endpoint_id и задайте ей значение идентификатора эндпоинта для источника;
        • target_endpoint_id и задайте ей значение идентификатора эндпоинта для приемника;

      • ресурс yandex_datatransfer_transfer.

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

      terraform validate
      

      Если в файлах конфигурации есть ошибки, Terraform на них укажет.

   3. Создайте необходимую инфраструктуру:

      1. Выполните команду для просмотра планируемых изменений:

         terraform plan
         

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

      2. Если вас устраивают планируемые изменения, внесите их:

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

            terraform apply
            

         2. Подтвердите изменение ресурсов.

         3. Дождитесь завершения операции.

      Трансфер активируется автоматически после создания.

Проверьте работоспособность трансфераПроверьте работоспособность трансфера

1. Дождитесь перехода трансфера в статус Реплицируется.

2. Убедитесь, что данные из потока Data Streams перенеслись в бакет Object Storage:

   1. В консоли управления выберите каталог, в котором находится бакет.
   2. В списке сервисов выберите Object Storage.
   3. Выберите бакет из списка.
   4. Проверьте, что бакет содержит файл <имя_потока>_0.raw (.json или .csv, в зависимости от выбранного выходного формата) с тестовыми данными.

3. Отправьте в поток Data Streams новое сообщение:

   {
       "device_id": "rhibbh3y08qm********",
       "datetime": "2020-06-06 09:49:54",
       "latitude": 55.71294467,
       "longitude": 37.66542005,
       "altitude": 429.13,
       "speed": 55.5,
       "battery_voltage": null,
       "cabin_temperature": 18,
       "fuel_level": 32
   }
   

4. Убедитесь, что в бакете Object Storage появились добавленные данные:

   1. В консоли управления выберите каталог, в котором находится бакет.
   2. В списке сервисов выберите Object Storage.
   3. Выберите бакет из списка.
   4. Проверьте, что в бакет добавился файл <имя_потока>_0-1_1.raw (.json или .csv, в зависимости от выбранного выходного формата) с новыми данными.

Удалите созданные ресурсыУдалите созданные ресурсы

Некоторые ресурсы платные. Чтобы за них не списывалась плата, удалите ресурсы, которые вы больше не будете использовать:

1. Удалите трансфер.
2. Удалите эндпоинты для источника и приемника.
3. Удалите объекты из бакета Object Storage.

Остальные ресурсы удалите в зависимости от способа их создания: