Yandex Cloud
  • Сервисы
  • Решения
  • Почему Yandex Cloud
  • Сообщество
  • Тарифы
  • Документация
  • Связаться с нами
Подключиться
Language / Region
© 2022 ООО «Яндекс.Облако»
Yandex Data Streams
  • Начало работы
    • Подготовка окружения
    • Создание потока данных
    • Сбор и поставка данных
      • Fluentd
      • Logstash
      • AWS CLI
    • Сохранение данных в ClickHouse
    • Обработка потока изменений из Debezium
  • Пошаговые инструкции
    • Все инструкции
    • Управление потоками данных
    • Работа с AWS CLI
      • Создание потока данных
      • Получение информации о потоке
      • Отправка данных в поток
      • Чтение данных из потока
      • Удаление потока данных
    • Работа с AWS SDK
      • Подготовка окружения
      • Создание потока данных
      • Отправка данных в поток
      • Чтение данных из потока
      • Удаление потока данных
  • Концепции
    • Обзор
    • Термины и определения
    • Cегменты и ключи сегментов
    • Квоты и лимиты
    • Сравнение с Yandex Message Queue
  • Практические руководства
    • Ввод данных в системы хранения
    • Хранение журналов работы приложения
    • Передача данных в микросервисных архитектурах
  • Управление доступом
  • Правила тарификации
  • HTTP API, совместимый с Amazon Kinesis Data Streams
    • Обзор
    • Методы
      • CreateStream
      • DecreaseStreamRetentionPeriod
      • DeleteStream
      • DescribeStream
      • IncreaseStreamRetentionPeriod
      • GetRecords
      • GetShardIterator
      • ListStreams
      • PutRecord
      • PutRecords
      • UpdateShardCount
    • Общие ошибки
    • Примеры
  • Вопросы и ответы
  1. Пошаговые инструкции
  2. Работа с AWS CLI
  3. Чтение данных из потока

Чтение данных из потока

Статья создана
Yandex.Cloud
    CLI

    Чтобы получить данные из потока из первого сегмента, выполните команду:

    SHARD_ITERATOR=$(aws kinesis get-shard-iterator \
      --endpoint <эндпоинт> \
      --shard-id shard-000001 \ 
      --shard-iterator-type LATEST \ 
      --stream-name <идентификатор_потока_данных> \ 
      --query 'ShardIterator'| tr -d \")
    aws kinesis get-records \
      --endpoint <эндпоинт> \
      --shard-iterator $SHARD_ITERATOR  
    
    • --endpoint — для чтения данных из потока по протоколу AWS Kinesis Data Streams укажите эндпоинт https://yds.serverless.yandexcloud.net.

    • --stream-name — состоит из зоны доступности, идентификатора каталога, идентификатора базы данных Yandex Managed Service for YDB и имени потока.

      Например, укажите идентификатор потока /ru-central1/aoeu1kuk2dhtaupdb1es/cc8029jgtuabequtgtbv/aws_stream, если:

      • aws_stream — имя потока;
      • ru-central1 — зона доступности;
      • aoeu1kuk2dhtaupdb1es — идентификатор каталога;
      • cc8029jgtuabequtgtbv — идентификатор базы данных YDB.

    Пример команды:

    SHARD_ITERATOR=$(aws kinesis get-shard-iterator \
      --endpoint https://yds.serverless.yandexcloud.net \ 
      --shard-id shard-000001 \
      --shard-iterator-type LATEST \ 
      --stream-name /ru-central1/aoeu1kuk2dhtaupdb1es/cc8029jgtuabequtgtbv/aws_stream 
      --query 'ShardIterator'| tr -d \")
    aws kinesis get-records \ 
      --endpoint https://yds.serverless.yandexcloud.net \  
      --shard-iterator $SHARD_ITERATOR
    

    Пример результата:

    {
      "Records": [
          {
              "SequenceNumber": "0",
              "Data": "eyJ1c2VyX2lkIjoidXNlcjEiLCJzY29yZSI6MTAwfQ==",
              "PartitionKey": "1"
          },
          {
              "SequenceNumber": "1",
              "Data": "eyJ1c2VyX2lkIjoidXNlcjEiLCJzY29yZSI6MTAwfQ==",
              "PartitionKey": "1"
          },
      ...
    }
    

    В примере выше данные в поле Data выводятся закодированными в кодировке BASE64, если данные декодировать, то будет выведена строка:

      {"user_id":"user1","score":100}
    

    Была ли статья полезна?

    Language / Region
    © 2022 ООО «Яндекс.Облако»