kda-flink-demo

Hands-on guide to use Apache Flink app in Kinesis Data Analytics to aggregate time-series data with tumbling & sliding window in real-time.

What is the architecture of Kinesis Data Analytics with Apache Flink?

What is a 'Tumbling Window'?

A tumbling windows assigner assigns each element to a window of a specified window size. Tumbling windows have a fixed size and do not overlap. For example, if you specify a tumbling window with a size of 5 minutes, the current window will be evaluated and a new window will be started every five minutes as illustrated by the following figure.

What does the incoming streaming data look like?

Will be generated by the send.py and sent to Kinesis Data Stream.

[
    {
      "type": "Feature",
      "properties": {
        "RECEIVED_ON": "2020-09-14T09:20:22.385",
        "N02_001": "14",
        "N02_002": "5",
        "N02_003": "北海道新幹線",
        "N02_004": "西日本旅客鉄道",
        "ID": "5_14",
        "COUNT": 20
      }
    },
    {
      "type": "Feature",
      "properties": {
        "RECEIVED_ON": "2020-09-14T09:22:25.325",
        "N02_001": "14",
        "N02_002": "5",
        "N02_003": "北海道新幹線",
        "N02_004": "西日本旅客鉄道",
        "ID": "5_14",
        "COUNT": 30
      }
    }
    ...
]

What does the aggregated output tumbling data look like?

RAILWAY_CLASS	RAILWAY_CLASS_COUNT	WINDOW_START	WINDOW_END
12	10	2020-09-14T09:22:00Z	2020-09-14T09:23:00Z
12	11	2020-09-14T09:23:00Z	2020-09-14T09:24:00Z
14	20	2020-09-14T09:22:00Z	2020-09-14T09:23:00Z
15	13	2020-09-14T09:23:00Z	2020-09-14T09:24:00Z

Hands-on Steps:

1. Create Aurora PostgreSQL Serverless Cluster

1-1. Create DB Subnet Group and Cluster

aws rds create-db-subnet-group \
--db-subnet-group-name database-vpc-subnet-group \
--db-subnet-group-description database_vpc_subnet_group \
--subnet-ids <subnet_id_1> <subnet_id_2> <subnet_id_3>

aws rds create-db-cluster --db-cluster-identifier sls-postgres \
--engine aurora-postgresql --engine-version 10.7 --engine-mode serverless \
--scaling-configuration MinCapacity=2,MaxCapacity=4 \
--enable-http-endpoint \
--master-username <username> --master-user-password <password> \
--enable-http-endpoint \
--db-subnet-group-name database-vpc-subnet-group \
--vpc-security-group-ids <security-group-id>

1-2. Create secret

aws secretsmanager create-secret --name sls-postgres-secret --secret-string "file://creds-sls-postgres.json"

1-3. Create Database 'monitoring'

$ aws rds-data execute-statement --resource-arn "<cluster-arn>" --secret-arn "<secret-arn>" --sql "SELECT datname FROM pg_database" --database "postgres"

$ aws rds-data execute-statement --resource-arn "<cluster-arn>" --secret-arn "<secret-arn>" --sql "CREATE DATABASE monitoring" --database "postgres"

1-4. Create Table 'Tumbling'

$ aws rds-data execute-statement --resource-arn "<cluster-arn>" --secret-arn "<secret-arn>" --sql "create table tumbling(RAILWAY_CLASS varchar(20), RAILWAY_CLASS_COUNT bigint, WINDOW_START timestamp, WINDOW_END timestamp, RECEIVED_ON timestamp, PRIMARY KEY(RAILWAY_CLASS, WINDOW_START, WINDOW_END))" --database "monitoring"

2. Create local flink-connector-kinesis repo

2-1. Download and Build Flink 1.8.2

curl -LJO https://github.com/apache/flink/archive/release-1.8.2.zip && unzip flink-release-1.8.2.zip && cd flink-release-1.8.2

mvn clean install -Pinclude-kinesis -DskipTests -pl flink-connectors/flink-connector-kinesis

2-2. Build this repo

The built shaded jar file is under the target folder.

mvn package -Dflink.version=1.8.2

3. Create and run the KDA application

Follow the AWS official document Create and Run the Application (Console)

3-1. Application Properties Groups

3-1-1. Group ID: JobConfigProperties

Key: "JOB_CLASS_NAME", Value: "StreamJobSqlTumbling"

3-1-2. Group ID: SourceConfigProperties

Key: "AWS_REGION", Value: "< your-aws-region >"
Key: "INPUT_STREAM_NAME", Value: "kda_geojson"
Key: "STREAM_INITIAL_POSITION", Value: "LATEST"

3-1-3. Group ID: SinkConfigProperties

Key: "DATABASE", Value: The database name set as 'monitoring' from 1-6
Key: "RESOURCE_ARN", Value: The datbase cluster ARN from 1-1
Key: "SECRET_ARN", Value: The secret ARN from 1-2
Key: "THRESHOLD", Value: 4

3-1-4. Group ID: ProcessorConfigProperties

Key: "CHECKPOINT_INTERVAL", Value: 30000
Key: "INTERVAL_AMOUNT", Value: 1
Key: "INTERVAL_UOM", Value: "MINUTE(2)"

3-2. Update Service execution IAM role

Default role name kinesis-analytics-flink-analytics-monitoring-< your-region >

3-2-1. Add inline policy 'read-all-secret' to read secrets

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": [
                "secretsmanager:GetSecretValue"
            ],
            "Resource": [
                "*"
            ],
            "Effect": "Allow"
        }
    ]
}

3-2-2. Add inline policy 'read-all-stream' to read streams

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "ReadInputStream",
            "Effect": "Allow",
            "Action": "kinesis:*",
            "Resource": "*"
        }
    ]
}

3-2-3. Add inline policy 'use-all-dataapi' to access Data API of Aurora Serverless

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": [
                "rds-data:BatchExecuteStatement",
                "rds-data:BeginTransaction",
                "rds-data:CommitTransaction",
                "rds-data:ExecuteStatement",
                "rds-data:RollbackTransaction",
                "rds-data:DeleteItems",
                "rds-data:ExecuteSql",
                "rds-data:GetItems",
                "rds-data:InsertItems",
                "rds-data:UpdateItems"
            ],
            "Resource": [
                "*"
            ],
            "Effect": "Allow"
        }
    ]
}