Example: Stagger Window

Sign in to the AWS Management Console and open the Kinesis console at https://console.aws.amazon.com/kinesis.

Choose Data Streams in the navigation pane.

Choose Create Kinesis stream, and then create a stream with one shard. For more information, see Create a Stream in the Amazon Kinesis Data Streams Developer Guide.

To write records to a Kinesis data stream in a production environment, we recommend using either the Kinesis Producer Library or Kinesis Data Streams API. For simplicity, this example uses the following Python script to generate records. Run the code to populate the sample ticker records. This simple code continuously writes a group of six records with the same random EVENT_TIME and ticker symbol to the stream, over the course of one minute. Keep the script running so that you can generate the application schema in a later step.

 
import datetime
import json
import random
import time
import boto3

STREAM_NAME = "ExampleInputStream"


def get_data():
    event_time = datetime.datetime.utcnow() - datetime.timedelta(seconds=10)
    return {
        "EVENT_TIME": event_time.isoformat(),
        "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
    }


def generate(stream_name, kinesis_client):
    while True:
        data = get_data()
        # Send six records, ten seconds apart, with the same event time and ticker
        for _ in range(6):
            print(data)
            kinesis_client.put_record(
                StreamName=stream_name,
                Data=json.dumps(data),
                PartitionKey="partitionkey",
            )
            time.sleep(10)


if __name__ == "__main__":
    generate(STREAM_NAME, boto3.client("kinesis"))

Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/kinesisanalytics.

Choose Create application, type an application name, and choose Create application.

On the application details page, choose Connect streaming data to connect to the source.

On the Connect to source page, do the following:

1. Choose the stream that you created in the preceding section.

2. Choose Discover Schema. Wait for the console to show the inferred schema and samples records that are used to infer the schema for the in-application stream created. The inferred schema has two columns.

3. Choose Edit Schema. Change the Column type of the EVENT_TIME column to TIMESTAMP.

4. Choose Save schema and update stream samples. After the console saves the schema, choose Exit.

5. Choose Save and continue.

On the application details page, choose Go to SQL editor. To start the application, choose Yes, start application in the dialog box that appears.

In the SQL editor, write the application code, and verify the results as follows:

1. Copy the following application code and paste it into the editor.

   CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
       event_time TIMESTAMP,
       ticker_symbol    VARCHAR(4),
       ticker_count     INTEGER);
   
   CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
     INSERT INTO "DESTINATION_SQL_STREAM" 
       SELECT STREAM 
           EVENT_TIME, 
           TICKER,
           COUNT(TICKER) AS ticker_count
       FROM "SOURCE_SQL_STREAM_001"
       WINDOWED BY STAGGER (
               PARTITION BY TICKER, EVENT_TIME RANGE INTERVAL '1' MINUTE);

2. Choose Save and run SQL.

   On the Real-time analytics tab, you can see all the in-application streams that the application created and verify the data.