Application

Source: src/AWS/KinesisAnalyticsV2/Application.ts

A Managed Service for Apache Flink (Kinesis Data Analytics v2) application.

Application owns the application definition — runtime environment, code location in S3, runtime properties, Flink settings, optional VPC connectivity and tags — and converges each aspect in place via UpdateApplication. Unless you supply serviceExecutionRole, an IAM role is auto-created granting the service read access to the code bucket and CloudWatch Logs delivery.

The application is created in READY and does not run (or bill KPUs) until started. Set start: true to have the reconciler start the job and wait for RUNNING — this requires the code object to be a real Flink application jar.

Creating Applications

Flink application from S3 code

import * as AWS from "alchemy/AWS";

const bucket = yield* AWS.S3.Bucket("FlinkCode");
const app = yield* AWS.KinesisAnalyticsV2.Application("Enrichment", {
  runtimeEnvironment: "FLINK-1_20",
  code: {
    bucketArn: bucket.bucketArn,
    fileKey: "jobs/enrichment-1.0.jar",
  },
});

Runtime properties and parallelism

const app = yield* AWS.KinesisAnalyticsV2.Application("Enrichment", {
  runtimeEnvironment: "FLINK-1_20",
  code: { bucketArn: bucket.bucketArn, fileKey: "jobs/enrichment-1.0.jar" },
  environmentProperties: [
    {
      propertyGroupId: "EnrichmentProperties",
      propertyMap: { "input.stream": "clickstream", "region": "us-west-2" },
    },
  ],
  flinkConfiguration: {
    parallelismConfiguration: {
      configurationType: "CUSTOM",
      parallelism: 2,
      parallelismPerKPU: 1,
      autoScalingEnabled: false,
    },
  },
  snapshotsEnabled: true,
});

VPC Connectivity

const app = yield* AWS.KinesisAnalyticsV2.Application("Enrichment", {
  runtimeEnvironment: "FLINK-1_20",
  code: { bucketArn: bucket.bucketArn, fileKey: "jobs/enrichment-1.0.jar" },
  vpc: {
    subnetIds: [subnetA.subnetId, subnetB.subnetId],
    securityGroupIds: [sg.securityGroupId],
  },
});

Maintenance Window

const app = yield* AWS.KinesisAnalyticsV2.Application("Enrichment", {
  runtimeEnvironment: "FLINK-1_20",
  code: { bucketArn: bucket.bucketArn, fileKey: "jobs/enrichment-1.0.jar" },
  maintenanceWindowStartTime: "02:00",
});

Running the Application

const app = yield* AWS.KinesisAnalyticsV2.Application("Enrichment", {
  runtimeEnvironment: "FLINK-1_20",
  code: { bucketArn: bucket.bucketArn, fileKey: "jobs/enrichment-1.0.jar" },
  start: true,
});