Потоковая обработка данных из Yandex Data Streams

В этом примере вы выполните потоковую обработку данных о поездках Нью-Йоркского такси. Данные для примера будут записаны с помощью генератора в специально созданный поток Yandex Data Streams.

В результате вы получите общую стоимость первых 10 поездок после начала обработки потоковых данных.

Для выполнения примера:

1. Подготовьтесь к работе.
2. Создайте поток данных.
3. Настройте генерацию данных.
4. Выполните запрос.
5. Исследуйте результат.

Подготовьтесь к работеПодготовьтесь к работе

1. Войдите в консоль управления или зарегистрируйтесь. Если вы еще не зарегистрированы, перейдите в консоль управления и следуйте инструкциям.
2. На странице Yandex Cloud Billing убедитесь, что у вас подключен платежный аккаунт и он находится в статусе ACTIVE или TRIAL_ACTIVE. Если платежного аккаунта нет, создайте его.
3. Если у вас еще нет каталога, создайте его.
4. Подключение к потоку данных будет выполняться от имени сервисного аккаунта. Создайте сервисный аккаунт с именем datastream-connection-account и ролью ydb.editor.
5. Потоки данных используют Yandex Managed Service for YDB. Создайте бессерверную базу данных.

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

1. В консоли управления выберите каталог, в котором нужно создать поток данных.
2. Перейдите в сервис Data Streams.
3. Нажмите кнопку Создать поток.
4. Укажите базу данных Yandex Managed Service for YDB, созданную ранее.
5. Введите имя потока данных: yellow-taxi.
6. Нажмите кнопку Создать.

Настройте генерацию данныхНастройте генерацию данных

1. Создайте соединение:

   1. В консоли управления выберите каталог, в котором нужно создать соединение.
   2. Перейдите в сервис Yandex Query.
   3. На панели слева выберите Учебник.
   4. Перейдите в Потоковый раздел.
   5. В блоке Создать инфраструктуру для обучения нажмите кнопку Создать соединение.
   6. В открывшемся окне в блоке Параметры типа соединения выберите базу данных и сервисный аккаунт, созданные ранее.
   7. Нажмите кнопку Создать.

2. Создайте привязку к данным:

   1. Откроется страница для создания привязки к данным.
   2. В блоке Параметры привязки к данным выберите поток yellow-taxi, созданный ранее.
   3. Нажмите кнопку Создать.

Запустится генерация данных в поток yellow-taxi. Для управления генератором данных используйте кнопки Стоп и Старт.

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

1. В редакторе запросов в интерфейсе Query нажмите кнопку Новый потоковый запрос.

2. В текстовом поле введите текст запроса:

   $data = 
   SELECT 
       *
   FROM
       bindings.`tutorial-streaming` LIMIT 10;
   
   SELECT
       HOP_END() AS time,
       COUNT(*) AS ride_count,
       SUM(total_amount) AS total_amount
   FROM 
       $data
   GROUP BY 
       HOP(CAST(tpep_pickup_datetime AS Timestamp), "PT1M", "PT1M", "PT1M");
   

3. Нажмите кнопку Выполнить.

Исследуйте результатИсследуйте результат

После выполнения запроса станет доступен результат — общая стоимость total_amount первых 10 поездок, совершенных после запуска запроса.

См. такжеСм. также

• Оператор HOP. Параметры окон в потоковой обработке данных
• Агрегатные функции. Синтаксис YQL
• Формат SQL-выражения
• Потоковый анализ данных