Unified streaming and batch data analysis

In this example, we will query analytical and streaming data to calculate taxi fares in specific locations.

We will use the same SQL query for both data types, with the only difference in bucket and stream connections and data bindings.

The data for batch processing has been pre-loaded in the Yandex Object Storage bucket in Parquet files. Streaming data will be written to a dedicated Yandex Data Streams stream by a generator.

In both cases, we will use a reference table stored in Object Storage to filter query data.

To run this example:

1. Make the necessary preparations.
2. Analyze the Object Storage data.
3. Analyze the streaming data from Data Streams.

Make the necessary preparationsMake the necessary preparations

1. Log in to the management console or sign up if you have not already. If you have not signed up yet, navigate to the management console and follow the instructions.
2. On the Yandex Cloud Billing page, make sure you have an ACTIVE or TRIAL_ACTIVE billing account. 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 the data stream using a service account. Create a service account named datastream-connection-account and assign it the ydb.editor role.
5. Data streams use Yandex Managed Service for YDB. You will need to create a serverless database.

Analyze the data Object StorageAnalyze the data Object Storage

Connect to the analytical data sourceConnect to the analytical data source

1. In the management console, select the folder where you want to create a connection.

2. Navigate to Yandex Query.

3. In the left-hand panel, select Tutorial.

4. Under Create infrastructure for tutorial, click Create connection.

   This will open the create connection page. Check the default settings; do not change them.

5. Click Create.

   This will open the create data binding page. Check the default settings; do not change them.

6. Click Create.

Run the queryRun the query

1. In the query editor within the Query interface, click New analytics query.

2. Enter the query text in the text field:

   $data =
   SELECT
       *
   FROM
       `tutorial-analytics`;
   
   $locations =
   SELECT
       PULocationID
   FROM
       `tutorial-analytics`.`nyc_taxi_sample/example_locations.csv`
   WITH
   (
       format=csv_with_names,
       SCHEMA
       (
           PULocationID String
       )
   );
   
   $time =
   SELECT
       HOP_END() AS time,
       rides.PULocationID AS PULocationID,
       SUM(total_amount) AS total_amount
   FROM $data AS rides
   INNER JOIN $locations AS locations
       ON rides.PULocationID=locations.PULocationID
   GROUP BY
       HOP(CAST(tpep_pickup_datetime AS Timestamp?), "PT1M", "PT1M", "PT1M"),
       rides.PULocationID;
   
   SELECT
       *
   FROM
       $time;
   

3. Click Run.

Check the resultCheck the result

Once executed, the analytical query will return the distribution of taxi trip fares in specific locations.

Analyze the streaming data from Data StreamsAnalyze the streaming data from Data Streams

Create a data streamCreate a data stream

1. In the management console, select the folder where you need to create a data stream.
2. Navigate to Data Streams.
3. Click Create stream.
4. Specify the Yandex Managed Service for YDB database created earlier.
5. Name the data 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. Navigate to Yandex Query.
   3. In the left-hand panel, select Tutorial.
   4. Navigate 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 you created earlier.
   7. Click Create.

2. Create a data binding:

   1. This will open the create data binding page.
   2. Under Binding parameters, select the yellow-taxi stream you created earlier.
   3. Click Create.

The generator will start writing data to the yellow-taxi stream. You can control the generator using the Stop and Start buttons.

Run the queryRun the query

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

2. Enter the query text in the text field:

   $data =
   SELECT
       *
   FROM bindings.`tutorial-streaming`;
   
   $locations =
   SELECT
       PULocationID
   FROM
       `tutorial-analytics`.`nyc_taxi_sample/example_locations.csv`
   WITH
   (
       format=csv_with_names,
       SCHEMA
       (
           PULocationID String
       )
   );
   
   $time =
   SELECT
       HOP_END() AS time,
       rides.PULocationID AS PULocationID,
       SUM(total_amount) AS total_amount
   FROM $data AS rides
   INNER JOIN $locations AS locations
       ON rides.PULocationID=locations.PULocationID
   GROUP BY
       HOP(cast(tpep_pickup_datetime AS Timestamp?), "PT1M", "PT1M", "PT1M"),
       rides.PULocationID;
   
   SELECT
       *
   FROM
       $time;
   

3. Click Run.

Check the resultCheck the result

Once launched, the query returns the total fare (total_amount) of the taxi rides taken in specific locations (PULocationID) after processing started.

See alsoSee also

• HOP operator and window parameters in streaming data processing
• Aggregate functions in YQL
• SQL syntax
• Batch processing
• Streaming data analysis