There are more AWS SDK examples available in the [AWS Doc SDK Examples](https://github.com/awsdocs/aws-doc-sdk-examples) GitHub repo.
# Actions for Managed Service for Apache Flink using AWS SDKs
The following code examples demonstrate how to perform individual Managed Service for Apache Flink actions with AWS SDKs. Each example includes a link to GitHub, where you can find instructions for setting up and running the code.
The following examples include only the most commonly used actions. For a complete list, see the [Amazon Managed Service for Apache Flink API Reference](https://docs.aws.amazon.com/managed-flink/latest/apiv2/Welcome.html).
**Topics**
+ [`AddApplicationInput`](kinesis-analytics-v2_example_kinesis-analytics-v2_AddApplicationInput_section.md)
+ [`AddApplicationOutput`](kinesis-analytics-v2_example_kinesis-analytics-v2_AddApplicationOutput_section.md)
+ [`CreateApplication`](kinesis-analytics-v2_example_kinesis-analytics-v2_CreateApplication_section.md)
+ [`DeleteApplication`](kinesis-analytics-v2_example_kinesis-analytics-v2_DeleteApplication_section.md)
+ [`DescribeApplication`](kinesis-analytics-v2_example_kinesis-analytics-v2_DescribeApplication_section.md)
+ [`DescribeApplicationSnapshot`](kinesis-analytics-v2_example_kinesis-analytics-v2_DescribeApplicationSnapshot_section.md)
+ [`DiscoverInputSchema`](kinesis-analytics-v2_example_kinesis-analytics-v2_DiscoverInputSchema_section.md)
+ [`StartApplication`](kinesis-analytics-v2_example_kinesis-analytics-v2_StartApplication_section.md)
+ [`StopApplication`](kinesis-analytics-v2_example_kinesis-analytics-v2_StopApplication_section.md)
+ [`UpdateApplication`](kinesis-analytics-v2_example_kinesis-analytics-v2_UpdateApplication_section.md)
# Use `AddApplicationInput` with an AWS SDK
The following code example shows how to use `AddApplicationInput`.
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis-analytics-v2#code-examples).
```
class KinesisAnalyticsApplicationV2:
"""Encapsulates Kinesis Data Analytics application functions."""
def __init__(self, analytics_client):
"""
:param analytics_client: A Boto3 Kinesis Data Analytics v2 client.
"""
self.analytics_client = analytics_client
self.name = None
self.arn = None
self.version_id = None
self.create_timestamp = None
def add_input(self, input_prefix, stream_arn, input_schema):
"""
Adds an input stream to the application. The input stream data is mapped
to an in-application stream that can be processed by your code running in
Kinesis Data Analytics.
:param input_prefix: The prefix prepended to in-application input stream names.
:param stream_arn: The ARN of the input stream.
:param input_schema: A schema that maps the data in the input stream to the
runtime environment. This can be automatically generated
by using `discover_input_schema` or you can create it
yourself.
:return: Metadata about the newly added input.
"""
try:
response = self.analytics_client.add_application_input(
ApplicationName=self.name,
CurrentApplicationVersionId=self.version_id,
Input={
"NamePrefix": input_prefix,
"KinesisStreamsInput": {"ResourceARN": stream_arn},
"InputSchema": input_schema,
},
)
self.version_id = response["ApplicationVersionId"]
logger.info("Add input stream %s to application %s.", stream_arn, self.name)
except ClientError:
logger.exception(
"Couldn't add input stream %s to application %s.", stream_arn, self.name
)
raise
else:
return response
```
+ For API details, see [AddApplicationInput](https://docs.aws.amazon.com/goto/boto3/kinesisanalyticsv2-2018-05-23/AddApplicationInput) in *AWS SDK for Python (Boto3) API Reference*.
------
# Use `AddApplicationOutput` with an AWS SDK
The following code example shows how to use `AddApplicationOutput`.
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis-analytics-v2#code-examples).
```
class KinesisAnalyticsApplicationV2:
"""Encapsulates Kinesis Data Analytics application functions."""
def __init__(self, analytics_client):
"""
:param analytics_client: A Boto3 Kinesis Data Analytics v2 client.
"""
self.analytics_client = analytics_client
self.name = None
self.arn = None
self.version_id = None
self.create_timestamp = None
def add_output(self, in_app_stream_name, output_arn):
"""
Adds an output stream to the application. Kinesis Data Analytics maps data
from the specified in-application stream to the output stream.
:param in_app_stream_name: The name of the in-application stream to map
to the output stream.
:param output_arn: The ARN of the output stream.
:return: A list of metadata about the output resources currently assigned
to the application.
"""
try:
response = self.analytics_client.add_application_output(
ApplicationName=self.name,
CurrentApplicationVersionId=self.version_id,
Output={
"Name": in_app_stream_name,
"KinesisStreamsOutput": {"ResourceARN": output_arn},
"DestinationSchema": {"RecordFormatType": "JSON"},
},
)
outputs = response["OutputDescriptions"]
self.version_id = response["ApplicationVersionId"]
logging.info(
"Added output %s to %s, which now has %s outputs.",
output_arn,
self.name,
len(outputs),
)
except ClientError:
logger.exception("Couldn't add output %s to %s.", output_arn, self.name)
raise
else:
return outputs
```
+ For API details, see [AddApplicationOutput](https://docs.aws.amazon.com/goto/boto3/kinesisanalyticsv2-2018-05-23/AddApplicationOutput) in *AWS SDK for Python (Boto3) API Reference*.
------
# Use `CreateApplication` with an AWS SDK
The following code example shows how to use `CreateApplication`.
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis-analytics-v2#code-examples).
```
class KinesisAnalyticsApplicationV2:
"""Encapsulates Kinesis Data Analytics application functions."""
def __init__(self, analytics_client):
"""
:param analytics_client: A Boto3 Kinesis Data Analytics v2 client.
"""
self.analytics_client = analytics_client
self.name = None
self.arn = None
self.version_id = None
self.create_timestamp = None
def create(self, app_name, role_arn, env="SQL-1_0"):
"""
Creates a Kinesis Data Analytics application.
:param app_name: The name of the application.
:param role_arn: The ARN of a role that can be assumed by Kinesis Data
Analytics and grants needed permissions.
:param env: The runtime environment of the application, such as SQL. Code
uploaded to the application runs in this environment.
:return: Metadata about the newly created application.
"""
try:
response = self.analytics_client.create_application(
ApplicationName=app_name,
RuntimeEnvironment=env,
ServiceExecutionRole=role_arn,
)
details = response["ApplicationDetail"]
self._update_details(details)
logger.info("Application %s created.", app_name)
except ClientError:
logger.exception("Couldn't create application %s.", app_name)
raise
else:
return details
```
+ For API details, see [CreateApplication](https://docs.aws.amazon.com/goto/boto3/kinesisanalyticsv2-2018-05-23/CreateApplication) in *AWS SDK for Python (Boto3) API Reference*.
------
# Use `DeleteApplication` with an AWS SDK
The following code example shows how to use `DeleteApplication`.
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis-analytics-v2#code-examples).
```
class KinesisAnalyticsApplicationV2:
"""Encapsulates Kinesis Data Analytics application functions."""
def __init__(self, analytics_client):
"""
:param analytics_client: A Boto3 Kinesis Data Analytics v2 client.
"""
self.analytics_client = analytics_client
self.name = None
self.arn = None
self.version_id = None
self.create_timestamp = None
def delete(self):
"""
Deletes an application.
"""
try:
self.analytics_client.delete_application(
ApplicationName=self.name, CreateTimestamp=self.create_timestamp
)
logger.info("Deleted application %s.", self.name)
except ClientError:
logger.exception("Couldn't delete application %s.", self.name)
raise
```
+ For API details, see [DeleteApplication](https://docs.aws.amazon.com/goto/boto3/kinesisanalyticsv2-2018-05-23/DeleteApplication) in *AWS SDK for Python (Boto3) API Reference*.
------
# Use `DescribeApplication` with an AWS SDK
The following code example shows how to use `DescribeApplication`.
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis-analytics-v2#code-examples).
```
class KinesisAnalyticsApplicationV2:
"""Encapsulates Kinesis Data Analytics application functions."""
def __init__(self, analytics_client):
"""
:param analytics_client: A Boto3 Kinesis Data Analytics v2 client.
"""
self.analytics_client = analytics_client
self.name = None
self.arn = None
self.version_id = None
self.create_timestamp = None
def describe(self, name):
"""
Gets metadata about an application.
:param name: The name of the application to look up.
:return: Metadata about the application.
"""
try:
response = self.analytics_client.describe_application(ApplicationName=name)
details = response["ApplicationDetail"]
self._update_details(details)
logger.info("Got metadata for application %s.", name)
except ClientError:
logger.exception("Couldn't get metadata for application %s.", name)
raise
else:
return details
```
+ For API details, see [DescribeApplication](https://docs.aws.amazon.com/goto/boto3/kinesisanalyticsv2-2018-05-23/DescribeApplication) in *AWS SDK for Python (Boto3) API Reference*.
------
# Use `DescribeApplicationSnapshot` with an AWS SDK
The following code example shows how to use `DescribeApplicationSnapshot`.
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis-analytics-v2#code-examples).
```
class KinesisAnalyticsApplicationV2:
"""Encapsulates Kinesis Data Analytics application functions."""
def __init__(self, analytics_client):
"""
:param analytics_client: A Boto3 Kinesis Data Analytics v2 client.
"""
self.analytics_client = analytics_client
self.name = None
self.arn = None
self.version_id = None
self.create_timestamp = None
def describe_snapshot(self, application_name, snapshot_name):
"""
Gets metadata about a previously saved application snapshot.
:param application_name: The name of the application.
:param snapshot_name: The name of the snapshot.
:return: Metadata about the snapshot.
"""
try:
response = self.analytics_client.describe_application_snapshot(
ApplicationName=application_name, SnapshotName=snapshot_name
)
snapshot = response["SnapshotDetails"]
logger.info(
"Got metadata for snapshot %s of application %s.",
snapshot_name,
application_name,
)
except ClientError:
logger.exception(
"Couldn't get metadata for snapshot %s of application %s.",
snapshot_name,
application_name,
)
raise
else:
return snapshot
```
+ For API details, see [DescribeApplicationSnapshot](https://docs.aws.amazon.com/goto/boto3/kinesisanalyticsv2-2018-05-23/DescribeApplicationSnapshot) in *AWS SDK for Python (Boto3) API Reference*.
------
# Use `DiscoverInputSchema` with an AWS SDK
The following code example shows how to use `DiscoverInputSchema`.
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis-analytics-v2#code-examples).
```
class KinesisAnalyticsApplicationV2:
"""Encapsulates Kinesis Data Analytics application functions."""
def __init__(self, analytics_client):
"""
:param analytics_client: A Boto3 Kinesis Data Analytics v2 client.
"""
self.analytics_client = analytics_client
self.name = None
self.arn = None
self.version_id = None
self.create_timestamp = None
def discover_input_schema(self, stream_arn, role_arn):
"""
Discovers a schema that maps data in a stream to a format that is usable by
an application's runtime environment. The stream must be active and have
enough data moving through it for the service to sample. The returned schema
can be used when you add the stream as an input to the application or you can
write your own schema.
:param stream_arn: The ARN of the stream to map.
:param role_arn: A role that lets Kinesis Data Analytics read from the stream.
:return: The discovered schema of the data in the input stream.
"""
try:
response = self.analytics_client.discover_input_schema(
ResourceARN=stream_arn,
ServiceExecutionRole=role_arn,
InputStartingPositionConfiguration={"InputStartingPosition": "NOW"},
)
schema = response["InputSchema"]
logger.info("Discovered input schema for stream %s.", stream_arn)
except ClientError:
logger.exception(
"Couldn't discover input schema for stream %s.", stream_arn
)
raise
else:
return schema
```
+ For API details, see [DiscoverInputSchema](https://docs.aws.amazon.com/goto/boto3/kinesisanalyticsv2-2018-05-23/DiscoverInputSchema) in *AWS SDK for Python (Boto3) API Reference*.
------
# Use `StartApplication` with an AWS SDK
The following code example shows how to use `StartApplication`.
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis-analytics-v2#code-examples).
```
class KinesisAnalyticsApplicationV2:
"""Encapsulates Kinesis Data Analytics application functions."""
def __init__(self, analytics_client):
"""
:param analytics_client: A Boto3 Kinesis Data Analytics v2 client.
"""
self.analytics_client = analytics_client
self.name = None
self.arn = None
self.version_id = None
self.create_timestamp = None
def start(self, input_id):
"""
Starts an application. After the application is running, it reads from the
specified input stream and runs the application code on the incoming data.
:param input_id: The ID of the input to read.
"""
try:
self.analytics_client.start_application(
ApplicationName=self.name,
RunConfiguration={
"SqlRunConfigurations": [
{
"InputId": input_id,
"InputStartingPositionConfiguration": {
"InputStartingPosition": "NOW"
},
}
]
},
)
logger.info("Started application %s.", self.name)
except ClientError:
logger.exception("Couldn't start application %s.", self.name)
raise
```
+ For API details, see [StartApplication](https://docs.aws.amazon.com/goto/boto3/kinesisanalyticsv2-2018-05-23/StartApplication) in *AWS SDK for Python (Boto3) API Reference*.
------
# Use `StopApplication` with an AWS SDK
The following code example shows how to use `StopApplication`.
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis-analytics-v2#code-examples).
```
class KinesisAnalyticsApplicationV2:
"""Encapsulates Kinesis Data Analytics application functions."""
def __init__(self, analytics_client):
"""
:param analytics_client: A Boto3 Kinesis Data Analytics v2 client.
"""
self.analytics_client = analytics_client
self.name = None
self.arn = None
self.version_id = None
self.create_timestamp = None
def stop(self):
"""
Stops an application. This stops the application from processing data but
does not delete any resources.
"""
try:
self.analytics_client.stop_application(ApplicationName=self.name)
logger.info("Stopping application %s.", self.name)
except ClientError:
logger.exception("Couldn't stop application %s.", self.name)
raise
```
+ For API details, see [StopApplication](https://docs.aws.amazon.com/goto/boto3/kinesisanalyticsv2-2018-05-23/StopApplication) in *AWS SDK for Python (Boto3) API Reference*.
------
# Use `UpdateApplication` with an AWS SDK
The following code example shows how to use `UpdateApplication`.
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/kinesis-analytics-v2#code-examples).
This example updates the code that runs in an existing application.
```
class KinesisAnalyticsApplicationV2:
"""Encapsulates Kinesis Data Analytics application functions."""
def __init__(self, analytics_client):
"""
:param analytics_client: A Boto3 Kinesis Data Analytics v2 client.
"""
self.analytics_client = analytics_client
self.name = None
self.arn = None
self.version_id = None
self.create_timestamp = None
def update_code(self, code):
"""
Updates the code that runs in the application. The code must run in the
runtime environment of the application, such as SQL. Application code
typically reads data from in-application streams and transforms it in some way.
:param code: The code to upload. This completely replaces any existing code
in the application.
:return: Metadata about the application.
"""
try:
response = self.analytics_client.update_application(
ApplicationName=self.name,
CurrentApplicationVersionId=self.version_id,
ApplicationConfigurationUpdate={
"ApplicationCodeConfigurationUpdate": {
"CodeContentTypeUpdate": "PLAINTEXT",
"CodeContentUpdate": {"TextContentUpdate": code},
}
},
)
details = response["ApplicationDetail"]
self.version_id = details["ApplicationVersionId"]
logger.info("Update code for application %s.", self.name)
except ClientError:
logger.exception("Couldn't update code for application %s.", self.name)
raise
else:
return details
```
+ For API details, see [UpdateApplication](https://docs.aws.amazon.com/goto/boto3/kinesisanalyticsv2-2018-05-23/UpdateApplication) in *AWS SDK for Python (Boto3) API Reference*.
------