Processing of data streams from Yandex Data Streams

In this example, you will process a data stream on New York City taxi rides. Data for the example will be written by a generator to a dedicated Yandex Data Streams stream.

As a result, you will get the total cost of the first ten rides since the stream data processing began.

To run this example:

1. Get things ready.
2. Create a data stream.
3. Set up data generation.
4. Run the query.
5. Review the result.

Get startedGet started

1. Log in or sign up to the management console. If not signed up yet, navigate to the management console and follow the instructions.
2. On the Yandex Cloud Billing page, make sure you have a billing account linked and it has the ACTIVE or TRIAL_ACTIVE status. If you do not have a billing account yet, create one.
3. If you do not have a folder yet, create one.
4. We will connect to our data stream using a service account. Create a service account named datastream-connection-account with the ydb.editor role.
5. Data streams use Yandex Managed Service for YDB. You will need to create a serverless database.

Create a data streamCreate a data stream

1. In the management console, select the folder where you need to create a data stream.
2. Select Data Streams.
3. Click Create stream.
4. Specify the Yandex Managed Service for YDB database created previously.
5. Enter the name of the stream: yellow-taxi.
6. Click Create.

Set up data generationSet up data generation

1. Create a connection.

   1. In the management console, select the folder where you want to create a connection.
   2. In the list of services, select Yandex Query.
   3. In the left-hand panel, select Tutorial.
   4. Go to Streaming.
   5. Under Create infrastructure for tutorial, click Create connection.
   6. In the window that opens, under Connection type parameters, select the database and service account that you created previously.
   7. Click Create.

2. Create a data binding:

   1. A page for creating a data binding will open.
   2. Under Binding parameters, select the yellow-taxi stream created previously.
   3. Click Create.

Data generation to the yellow-taxi stream will start. Use the Stop and Start buttons to control the data generator.

Run the queryRun the query

1. In the query editor in the Query interface, click New streaming query.

2. Enter the query text in the text field:

   $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. Click Run.

Review the resultReview the result

Once the query is completed, you will see the result with the total cost (total_amount) of the first 10 rides taken after running the query.

See alsoSee also

• HOP. Window parameters in streamed data processing
• Aggregate functions. YQL syntax
• SQL expression format
• Streaming data analysis