Skip to main content

The CDK Construct Library for AWS::Kinesis

Project description

Amazon Kinesis Construct Library


cfn-resources: Stable

cdk-constructs: Stable

Amazon Kinesis provides collection and processing of large streams of data records in real time. Kinesis data streams can be used for rapid and continuous data intake and aggregation.

Table Of Contents


Amazon Kinesis Data Streams ingests a large amount of data in real time, durably stores the data, and makes the data available for consumption.

Using the CDK, a new Kinesis stream can be created as part of the stack using the construct's constructor. You may specify the streamName to give your own identifier to the stream. If not, CloudFormation will generate a name.

kinesis.Stream(self, "MyFirstStream",

You can also specify properties such as shardCount to indicate how many shards the stream should choose and a retentionPeriod to specify how long the data in the shards should remain accessible. Read more at Creating and Managing Streams

kinesis.Stream(self, "MyFirstStream",


Stream encryption enables server-side encryption using an AWS KMS key for a specified stream.

Encryption is enabled by default on your stream with the master key owned by Kinesis Data Streams in regions where it is supported.

kinesis.Stream(self, "MyEncryptedStream")

You can enable encryption on your stream with a user-managed key by specifying the encryption property. A KMS key will be created for you and associated with the stream.

kinesis.Stream(self, "MyEncryptedStream",

You can also supply your own external KMS key to use for stream encryption by specifying the encryptionKey property.

key = kms.Key(self, "MyKey")

kinesis.Stream(self, "MyEncryptedStream",


Any Kinesis stream that has been created outside the stack can be imported into your CDK app.

Streams can be imported by their ARN via the Stream.fromStreamArn() API

imported_stream = kinesis.Stream.from_stream_arn(self, "ImportedStream", "arn:aws:kinesis:us-east-2:123456789012:stream/f3j09j2230j")

Encrypted Streams can also be imported by their attributes via the Stream.fromStreamAttributes() API

imported_stream = kinesis.Stream.from_stream_attributes(self, "ImportedEncryptedStream",
    encryption_key=kms.Key.from_key_arn(self, "key", "arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012")

Permission Grants

IAM roles, users or groups which need to be able to work with Amazon Kinesis streams at runtime should be granted IAM permissions.

Any object that implements the IGrantable interface (has an associated principal) can be granted permissions by calling:

  • grantRead(principal) - grants the principal read access
  • grantWrite(principal) - grants the principal write permissions to a Stream
  • grantReadWrite(principal) - grants principal read and write permissions

Read Permissions

Grant read access to a stream by calling the grantRead() API. If the stream has an encryption key, read permissions will also be granted to the key.

lambda_role = iam.Role(self, "Role",
    description="Example role..."

stream = kinesis.Stream(self, "MyEncryptedStream",

# give lambda permissions to read stream

The following read permissions are provided to a service principal by the grantRead() API:

  • kinesis:DescribeStreamSummary
  • kinesis:GetRecords
  • kinesis:GetShardIterator
  • kinesis:ListShards
  • kinesis:SubscribeToShard

Write Permissions

Grant write permissions to a stream is provided by calling the grantWrite() API. If the stream has an encryption key, write permissions will also be granted to the key.

lambda_role = iam.Role(self, "Role",
    description="Example role..."

stream = kinesis.Stream(self, "MyEncryptedStream",

# give lambda permissions to write to stream

The following write permissions are provided to a service principal by the grantWrite() API:

  • kinesis:ListShards
  • kinesis:PutRecord
  • kinesis:PutRecords

Custom Permissions

You can add any set of permissions to a stream by calling the grant() API.

user = iam.User(self, "MyUser")

stream = kinesis.Stream(self, "MyStream")

# give my user permissions to list shards
stream.grant(user, "kinesis:ListShards")


You can use common metrics from your stream to create alarms and/or dashboards. The stream.metric('MetricName') method creates a metric with the stream namespace and dimension. You can also use pre-define methods like stream.metricGetRecordsSuccess(). To find out more about Kinesis metrics check Monitoring the Amazon Kinesis Data Streams Service with Amazon CloudWatch.

stream = kinesis.Stream(self, "MyStream")

# Using base metric method passing the metric name

# using pre-defined metric method

# using pre-defined and overriding the statistic

Project details

Release history Release notifications | RSS feed

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for, version 1.134.0
Filename, size File type Python version Upload date Hashes
Filename, size aws_cdk.aws_kinesis-1.134.0-py3-none-any.whl (72.4 kB) File type Wheel Python version py3 Upload date Hashes View
Filename, size (73.7 kB) File type Source Python version None Upload date Hashes View

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor NVIDIA NVIDIA PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page