View a markdown version of this page

Managed Service for Apache Flink API example code - Managed Service for Apache Flink
AddApplicationCloudWatchLoggingOptionAddApplicationInputAddApplicationInputProcessingConfigurationAddApplicationOutputAddApplicationReferenceDataSourceAddApplicationVpcConfigurationCreateApplicationCreateApplicationSnapshotDeleteApplicationDeleteApplicationCloudWatchLoggingOptionDeleteApplicationInputProcessingConfigurationDeleteApplicationOutputDeleteApplicationReferenceDataSourceCreateSnapshotDeleteApplicationSnapshotDeleteApplicationVpcConfigurationDeleteSnapshotDescribeApplicationDescribeApplicationSnapshotDescribeSnapshotDescribeSnapshotOperationDiscoverInputSchemaExportSnapshotImportSnapshotListApplicationsListApplicationSnapshotsListSnapshotOperationsListSnapshotsStartApplicationStopApplicationUpdateApplicationEnhanced snapshot APIs

Managed Service for Apache Flink API example code

This topic contains example request blocks for Managed Service for Apache Flink actions.

To use JSON as the input for an action with the AWS Command Line Interface (AWS CLI), save the request in a JSON file. Then pass the file name into the action using the --cli-input-json parameter.

The following example demonstrates how to use a JSON file with an action.

$ aws kinesisanalyticsv2 start-application --cli-input-json file://start.json

For more information about using JSON with the AWS CLI, see Generate CLI Skeleton and CLI Input JSON Parameters in the AWS Command Line Interface User Guide.

AddApplicationCloudWatchLoggingOption

The following example request code for the AddApplicationCloudWatchLoggingOption action adds an Amazon CloudWatch logging option to a Managed Service for Apache Flink application:

{
    "ApplicationName": "MyApplication",
    "CloudWatchLoggingOption": {
        "LogStreamARN": "arn:aws:logs:us-east-1:123456789123:log-group:my-log-group:log-stream:My-LogStream"
    },
    "CurrentApplicationVersionId": 2
}

AddApplicationInput

The following example request code for the AddApplicationInput action adds an application input to a Managed Service for Apache Flink application:

{
   "ApplicationName": "MyApplication",
   "CurrentApplicationVersionId": 2,
   "Input": { 
      "InputParallelism": { 
         "Count": 2
      },
      "InputSchema": { 
         "RecordColumns": [ 
            { 
               "Mapping": "$.TICKER",
               "Name": "TICKER_SYMBOL",
               "SqlType": "VARCHAR(50)"
            },
            {
                "SqlType": "REAL",
                "Name": "PRICE",
                "Mapping": "$.PRICE"
            }
         ],
         "RecordEncoding": "UTF-8",
         "RecordFormat": { 
            "MappingParameters": { 
               "JSONMappingParameters": { 
                  "RecordRowPath": "$"
               }
            },
            "RecordFormatType": "JSON"
         }
      },
      "KinesisStreamsInput": { 
         "ResourceARN": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleInputStream"
      }
   }
}

AddApplicationInputProcessingConfiguration

The following example request code for the AddApplicationInputProcessingConfiguration action adds an application input processing configuration to a Managed Service for Apache Flink application:

{
   "ApplicationName": "MyApplication",
   "CurrentApplicationVersionId": 2,
   "InputId": "2.1",
   "InputProcessingConfiguration": { 
      "InputLambdaProcessor": { 
         "ResourceARN": "arn:aws:lambda:us-east-1:012345678901:function:MyLambdaFunction"
      }
   }
}

AddApplicationOutput

The following example request code for the AddApplicationOutput action adds a Kinesis data stream as an application output to a Managed Service for Apache Flink application:

{
   "ApplicationName": "MyApplication",
   "CurrentApplicationVersionId": 2,
   "Output": { 
      "DestinationSchema": { 
         "RecordFormatType": "JSON"
      },
      "KinesisStreamsOutput": { 
         "ResourceARN": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleOutputStream"
      },
      "Name": "DESTINATION_SQL_STREAM"
   }
}

AddApplicationReferenceDataSource

The following example request code for the AddApplicationReferenceDataSource action adds a CSV application reference data source to a Managed Service for Apache Flink application:

{
   "ApplicationName": "MyApplication",
   "CurrentApplicationVersionId": 5,
   "ReferenceDataSource": { 
      "ReferenceSchema": { 
         "RecordColumns": [ 
            { 
               "Mapping": "$.TICKER",
               "Name": "TICKER",
               "SqlType": "VARCHAR(4)"
            },
            { 
               "Mapping": "$.COMPANYNAME",
               "Name": "COMPANY_NAME",
               "SqlType": "VARCHAR(40)"
            },
         ],
         "RecordEncoding": "UTF-8",
         "RecordFormat": { 
            "MappingParameters": { 
               "CSVMappingParameters": { 
                  "RecordColumnDelimiter": " ",
                  "RecordRowDelimiter": "\r\n"
               }
            },
            "RecordFormatType": "CSV"
         }
      },
      "S3ReferenceDataSource": { 
         "BucketARN": "arn:aws:s3:::amzn-s3-demo-bucket",
         "FileKey": "TickerReference.csv"
      },
      "TableName": "string"
   }
}

AddApplicationVpcConfiguration

The following example request code for the AddApplicationVpcConfiguration action adds a VPC configuration to an existing application:

{
   "ApplicationName": "MyApplication",
   "CurrentApplicationVersionId": 9,
   "VpcConfiguration": { 
      "SecurityGroupIds": [ "sg-0123456789abcdef0" ],
      "SubnetIds": [ "subnet-0123456789abcdef0" ]
   }
}

CreateApplication

The following example request code for the CreateApplication action creates a Managed Service for Apache Flink application:

{
  "ApplicationName":"MyApplication",
  "ApplicationDescription":"My-Application-Description",
  "RuntimeEnvironment":"FLINK-1_15",
  "ServiceExecutionRole":"arn:aws:iam::123456789123:role/myrole",
  "CloudWatchLoggingOptions":[
    {
      "LogStreamARN":"arn:aws:logs:us-east-1:123456789123:log-group:my-log-group:log-stream:My-LogStream"
    }
  ],
  "ApplicationConfiguration": {
    "EnvironmentProperties":
      {"PropertyGroups": 
        [
          {"PropertyGroupId": "ConsumerConfigProperties",
            "PropertyMap": 
              {"aws.region": "us-east-1",
              "flink.stream.initpos": "LATEST"}
          },
          {"PropertyGroupId": "ProducerConfigProperties",
            "PropertyMap": 
              {"aws.region": "us-east-1"}
          },
        ]
      },
    "ApplicationCodeConfiguration":{
      "CodeContent":{
        "S3ContentLocation":{
          "BucketARN":"arn:aws:s3:::amzn-s3-demo-bucket",
          "FileKey":"myflink.jar",
          "ObjectVersion":"AbCdEfGhIjKlMnOpQrStUvWxYz12345"
        }
      },
      "CodeContentType":"ZIPFILE"
    },
      "FlinkApplicationConfiguration":{
      "ParallelismConfiguration":{
        "ConfigurationType":"CUSTOM",
        "Parallelism":2,
        "ParallelismPerKPU":1,
        "AutoScalingEnabled":true
      }
    }
  }
}

The following JSON example request for the CreateApplication action creates an application in enhanced snapshot mode. Set SnapshotMode to ENHANCED_SNAPSHOTS in ApplicationSnapshotConfiguration. If you omit SnapshotMode, the application uses APPLICATION_SNAPSHOTS_LEGACY. Enhanced snapshot mode requires Apache Flink version 1.20 or later.

In the following example, replace MyApplication, the ServiceExecutionRole ARN, the bucket ARN, and the file key with your own values.

{
    "ApplicationName": "MyApplication",
    "RuntimeEnvironment": "FLINK-1_20",
    "ServiceExecutionRole": "arn:aws:iam::111122223333:role/MyApplicationServiceRole",
    "ApplicationConfiguration": {
        "ApplicationCodeConfiguration": {
            "CodeContent": {
                "S3ContentLocation": {
                    "BucketARN": "arn:aws:s3:::amzn-s3-demo-bucket",
                    "FileKey": "my-flink-app.jar"
                }
            },
            "CodeContentType": "ZIPFILE"
        },
        "ApplicationSnapshotConfiguration": {
            "SnapshotsEnabled": true,
            "SnapshotMode": "ENHANCED_SNAPSHOTS"
        }
    }
}

CreateApplicationSnapshot

The following JSON example request for the CreateApplicationSnapshot action creates an application snapshot (legacy). For an application in enhanced snapshot mode, this action returns InvalidApplicationConfigurationException. For the response syntax, see CreateApplicationSnapshot in the Managed Service for Apache Flink API Reference.

{
   "ApplicationName": "MyApplication",
   "SnapshotName": "MySnapshot"
}

DeleteApplication

The following example request code for the DeleteApplication action deletes a Managed Service for Apache Flink application:

{"ApplicationName": "MyApplication",
"CreateTimestamp": 12345678912}

In enhanced snapshot mode, DeleteApplication does not delete the enhanced snapshots of the application. For more information, see Delete an enhanced snapshot.

DeleteApplicationCloudWatchLoggingOption

The following example request code for the DeleteApplicationCloudWatchLoggingOption action deletes an Amazon CloudWatch logging option from a Managed Service for Apache Flink application:

{
    "ApplicationName": "MyApplication",
    "CloudWatchLoggingOptionId": "3.1"
    "CurrentApplicationVersionId": 3
}

DeleteApplicationInputProcessingConfiguration

The following example request code for the DeleteApplicationInputProcessingConfiguration action removes an input processing configuration from a Managed Service for Apache Flink application:

{
   "ApplicationName": "MyApplication",
   "CurrentApplicationVersionId": 4,
   "InputId": "2.1"
}

DeleteApplicationOutput

The following example request code for the DeleteApplicationOutput action removes an application output from a Managed Service for Apache Flink application:

{
   "ApplicationName": "MyApplication",
   "CurrentApplicationVersionId": 4,
   "OutputId": "4.1"
}

DeleteApplicationReferenceDataSource

The following example request code for the DeleteApplicationReferenceDataSource action removes an application reference data source from a Managed Service for Apache Flink application:

{
   "ApplicationName": "MyApplication",
   "CurrentApplicationVersionId": 5,
   "ReferenceId": "5.1"
}

CreateSnapshot

The following JSON example request for the CreateSnapshot action creates an enhanced snapshot of a running application. The application must use enhanced snapshot mode.

{
   "ApplicationName": "MyApplication",
   "SnapshotName": "MyApplication-snapshot-1",
   "SnapshotDescription": "Before code update"
}

The following example shows the response:

{
   "SnapshotArn": "arn:aws:kinesisanalytics:aa-example-1:111122223333:snapshot/MyApplication-snapshot-1",
   "SnapshotStatus": "CREATING",
   "SnapshotOperationId": "a1B2c3D4e5F6",
   "SnapshotOperationStatus": "IN_PROGRESS"
}

DeleteApplicationSnapshot

The following JSON example request for the DeleteApplicationSnapshot action deletes an application snapshot (legacy). For an enhanced snapshot, use DeleteSnapshot. For the response syntax, see DeleteApplicationSnapshot in the Managed Service for Apache Flink API Reference.

{
   "ApplicationName": "MyApplication",
   "SnapshotCreationTimestamp": 12345678912,
   "SnapshotName": "MySnapshot"
}

DeleteApplicationVpcConfiguration

The following example request code for the DeleteApplicationVpcConfiguration action removes an existing VPC configuration from an application:

{
   "ApplicationName": "MyApplication",
   "CurrentApplicationVersionId": 9,
   "VpcConfigurationId": "1.1"
}

DeleteSnapshot

The following JSON example request for the DeleteSnapshot action deletes an enhanced snapshot. Copy the SnapshotCreationTimestamp value from the DescribeSnapshot response.

{
   "SnapshotName": "MyApplication-snapshot-1",
   "SnapshotCreationTimestamp": 1767225600
}

The following example shows the response:

{
   "SnapshotArn": "arn:aws:kinesisanalytics:aa-example-1:111122223333:snapshot/MyApplication-snapshot-1",
   "SnapshotStatus": "DELETING",
   "SnapshotOperationId": "g7H8i9J0k1L2",
   "SnapshotOperationStatus": "IN_PROGRESS"
}

DescribeApplication

The following example request code for the DescribeApplication action returns details about a Managed Service for Apache Flink application:

{"ApplicationName": "MyApplication"}

DescribeApplicationSnapshot

The following JSON example request for the DescribeApplicationSnapshot action returns details about an application snapshot (legacy). For an enhanced snapshot, use DescribeSnapshot. For the response syntax, see DescribeApplicationSnapshot in the Managed Service for Apache Flink API Reference.

{
   "ApplicationName": "MyApplication",
   "SnapshotName": "MySnapshot"
}

DescribeSnapshot

The following JSON example request for the DescribeSnapshot action returns the details of an enhanced snapshot.

{
   "SnapshotName": "MyApplication-snapshot-1"
}

The following example shows part of the response:

{
   "SnapshotInfoDetails": {
      "SnapshotName": "MyApplication-snapshot-1",
      "SnapshotArn": "arn:aws:kinesisanalytics:aa-example-1:111122223333:snapshot/MyApplication-snapshot-1",
      "SnapshotStatus": "READY",
      "SnapshotCreationTimestamp": 1767225600,
      "RuntimeEnvironment": "FLINK-1_20",
      "SnapshotInitiator": "USER",
      "SnapshotInitiatorOperation": "CreateSnapshot",
      "SnapshotFormat": "CANONICAL"
   }
}

DescribeSnapshotOperation

The following JSON example request for the DescribeSnapshotOperation action returns the details of one operation on an enhanced snapshot.

{
   "SnapshotName": "MyApplication-snapshot-1",
   "SnapshotOperationId": "s9T0u1V2w3X4"
}

The following example shows the response for a successful export:

{
   "SnapshotOperationInfoDetails": {
      "SnapshotOperationId": "s9T0u1V2w3X4",
      "SnapshotOperation": "ExportSnapshot",
      "SnapshotOperationStatus": "SUCCESSFUL",
      "SnapshotArn": "arn:aws:kinesisanalytics:aa-example-1:111122223333:snapshot/MyApplication-snapshot-1",
      "StartTime": 1767229200,
      "EndTime": 1767229800
   }
}

DiscoverInputSchema

The following example request code for the DiscoverInputSchema action generates a schema from a streaming source:

{
   "InputProcessingConfiguration": { 
      "InputLambdaProcessor": { 
         "ResourceARN": "arn:aws:lambda:us-east-1:012345678901:function:MyLambdaFunction"
      }
   },
   "InputStartingPositionConfiguration": { 
      "InputStartingPosition": "NOW"
   },
   "ResourceARN": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleInputStream",
   "S3Configuration": { 
      "BucketARN": "string",
      "FileKey": "string"
   },
   "ServiceExecutionRole": "string"
}

The following example request code for the DiscoverInputSchema action generates a schema from a reference source:

{
   "S3Configuration": { 
      "BucketARN": "arn:aws:s3:::amzn-s3-demo-bucket",
      "FileKey": "TickerReference.csv"
   },
   "ServiceExecutionRole": "arn:aws:iam::123456789123:role/myrole"
}

ExportSnapshot

The following JSON example request for the ExportSnapshot action exports an enhanced snapshot to an Amazon S3 bucket. For the role and bucket requirements, see Import and export enhanced snapshots.

In the following example, replace the snapshot name, the role ARN, the bucket ARN, and the key prefix with your own values.

{
   "SnapshotName": "MyApplication-snapshot-1",
   "ExportDestinationConfiguration": {
      "DestinationType": "S3",
      "ExportRoleArn": "arn:aws:iam::111122223333:role/MSFSnapshotExportRole",
      "DestinationEncryptionConfiguration": {
         "KeyType": "AWS_OWNED_KEY"
      },
      "DestinationS3PrefixLocation": {
         "BucketARN": "arn:aws:s3:::amzn-s3-demo-bucket",
         "KeyPrefix": "exports/my-app/"
      }
   }
}

The following example shows the response:

{
   "SnapshotOperationId": "s9T0u1V2w3X4",
   "SnapshotOperationStatus": "IN_PROGRESS"
}

ImportSnapshot

The following JSON example request for the ImportSnapshot action creates a new enhanced snapshot from the state files in an Amazon S3 prefix. For the file, role, and bucket requirements, see Import and export enhanced snapshots.

In the following example, replace the snapshot name, the role ARN, the bucket ARN, and the key prefix with your own values.

{
   "SnapshotName": "MyApplication-snapshot-imported",
   "ImportSourceConfiguration": {
      "SourceType": "S3",
      "ImportRoleArn": "arn:aws:iam::111122223333:role/MSFSnapshotImportRole",
      "SourceS3PrefixLocation": {
         "BucketARN": "arn:aws:s3:::amzn-s3-demo-bucket",
         "KeyPrefix": "imports/my-app/"
      }
   },
   "SnapshotDescription": "Imported from self-managed Apache Flink"
}

The following example shows the response:

{
   "SnapshotArn": "arn:aws:kinesisanalytics:aa-example-1:111122223333:snapshot/MyApplication-snapshot-imported",
   "SnapshotStatus": "CREATING",
   "SnapshotOperationId": "m3N4o5P6q7R8",
   "SnapshotOperationStatus": "IN_PROGRESS"
}

ListApplications

The following example request code for the ListApplications action returns a list of Managed Service for Apache Flink applications in your account:

{
   "ExclusiveStartApplicationName": "MyApplication",
   "Limit": 50
}

ListApplicationSnapshots

The following JSON example request for the ListApplicationSnapshots action returns a list of the application snapshots (legacy) of an application. For enhanced snapshots, use ListSnapshots. For the response syntax, see ListApplicationSnapshots in the Managed Service for Apache Flink API Reference.

{"ApplicationName": "MyApplication",
   "Limit": 50,
   "NextToken": "aBcDeFgHiJkLmNoPqRsTuVwXyZ0123"
}

ListSnapshotOperations

The following JSON example request for the ListSnapshotOperations action lists the failed operations of an enhanced snapshot.

{
   "SnapshotName": "MyApplication-snapshot-1",
   "SnapshotOperationStatus": "FAILED",
   "Limit": 50
}

The following example shows the response:

{
   "SnapshotOperationInfoList": [
      {
         "SnapshotOperationId": "y5Z6a7B8c9D0",
         "SnapshotOperation": "ExportSnapshot",
         "SnapshotOperationStatus": "FAILED",
         "StartTime": 1767232800,
         "EndTime": 1767233100
      }
   ]
}

ListSnapshots

The following JSON example request for the ListSnapshots action lists the first 50 enhanced snapshots in the account and Region.

{
   "Limit": 50
}

The following example shows part of the response:

{
   "SnapshotInfoList": [
      {
         "SnapshotName": "MyApplication-snapshot-1",
         "SnapshotArn": "arn:aws:kinesisanalytics:aa-example-1:111122223333:snapshot/MyApplication-snapshot-1",
         "SnapshotStatus": "READY",
         "SnapshotCreationTimestamp": 1767225600,
         "SnapshotInitiator": "USER",
         "SnapshotInitiatorOperation": "CreateSnapshot",
         "SourceApplicationName": "MyApplication",
         "SnapshotFormat": "CANONICAL"
      }
   ]
}

StartApplication

The following example request code for the StartApplication action starts a Managed Service for Apache Flink application, and loads the application state from the latest snapshot (if any):

{
    "ApplicationName": "MyApplication",
    "RunConfiguration": {
        "ApplicationRestoreConfiguration": { 
         "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
         }
    }
}

In enhanced snapshot mode, this request restores from the newest enhanced snapshot in the READY status that the application created. The request does not use an application snapshot (legacy). The caller needs the kinesisanalytics:RestoreFromSnapshot IAM permission on that enhanced snapshot. For more information, see Restore from the latest snapshot and IAM permissions.

The following JSON example request for the StartApplication action starts an application in enhanced snapshot mode from a specific enhanced snapshot. Set ApplicationRestoreType to RESTORE_FROM_CUSTOM_SNAPSHOT and identify the enhanced snapshot with SnapshotArn. Use SnapshotName for an application snapshot (legacy), and use SnapshotArn for an enhanced snapshot. The caller needs the kinesisanalytics:RestoreFromSnapshot IAM permission on the enhanced snapshot. For more information, see IAM permissions.

In the following example, replace MyApplication and the SnapshotArn value with your application name and the ARN of your enhanced snapshot.

{
    "ApplicationName": "MyApplication",
    "RunConfiguration": {
        "ApplicationRestoreConfiguration": {
            "ApplicationRestoreType": "RESTORE_FROM_CUSTOM_SNAPSHOT",
            "SnapshotArn": "arn:aws:kinesisanalytics:aa-example-1:111122223333:snapshot/MyApplication-snapshot-1"
        }
    }
}

StopApplication

The following example request code for the API_StopApplication action stops a Managed Service for Apache Flink application:

{"ApplicationName": "MyApplication"}

UpdateApplication

The following example request code for the UpdateApplication action updates a Managed Service for Apache Flink application to change the location of the application code:

{"ApplicationName": "MyApplication",
"CurrentApplicationVersionId": 1,
"ApplicationConfigurationUpdate": { 
      "ApplicationCodeConfigurationUpdate": { 
         "CodeContentTypeUpdate": "ZIPFILE",
         "CodeContentUpdate": { 
            "S3ContentLocationUpdate": { 
               "BucketARNUpdate": "arn:aws:s3:::amzn-s3-demo-bucket",
               "FileKeyUpdate": "my_new_code.zip",
               "ObjectVersionUpdate": "2"
         }
      }
   }
}

The following JSON example request for the UpdateApplication action upgrades an existing application to enhanced snapshot mode. The request must include SnapshotsEnabledUpdate together with SnapshotModeUpdate.

Note

You cannot set SnapshotModeUpdate back to APPLICATION_SNAPSHOTS_LEGACY. For more information, see Upgrade an existing application to enhanced snapshots.

The application must already use Apache Flink 1.20 or later. If you upgrade an application to enhanced snapshot mode in the READY status, the next start with RESTORE_FROM_LATEST_SNAPSHOT starts the application without state. To keep the state, upgrade the application to enhanced snapshot mode while it is RUNNING.

In the following example, replace MyApplication with your application name. Set CurrentApplicationVersionId to the current version ID from the DescribeApplication response.

{
    "ApplicationName": "MyApplication",
    "CurrentApplicationVersionId": 3,
    "ApplicationConfigurationUpdate": {
        "ApplicationSnapshotConfigurationUpdate": {
            "SnapshotsEnabledUpdate": true,
            "SnapshotModeUpdate": "ENHANCED_SNAPSHOTS"
        }
    }
}

The following JSON example request for the UpdateApplication action restores a running application in enhanced snapshot mode from a specific enhanced snapshot. The application must be in the RUNNING status. The caller needs kinesisanalytics:RestoreFromSnapshot on the enhanced snapshot.

In the following example, replace MyApplication, the CurrentApplicationVersionId value, and the SnapshotArn value with your own values.

{
    "ApplicationName": "MyApplication",
    "CurrentApplicationVersionId": 4,
    "RunConfigurationUpdate": {
        "ApplicationRestoreConfiguration": {
            "ApplicationRestoreType": "RESTORE_FROM_CUSTOM_SNAPSHOT",
            "SnapshotArn": "arn:aws:kinesisanalytics:aa-example-1:111122223333:snapshot/MyApplication-snapshot-1"
        }
    }
}

Enhanced snapshot APIs

Enhanced snapshots use different API actions than application snapshots (legacy). For sample requests and responses for these actions, see Enhanced snapshot API examples.