Getting started with streaming ingestion from Amazon Kinesis Data Streams

Create an IAM role with a trust policy that allows your Amazon Redshift cluster or Amazon Redshift Serverless workgroup to assume the role. For information about how to configure the trust policy for the IAM role, see Authorizing Amazon Redshift to access other Amazon services on your behalf. After it is created, the role should have the following IAM policy, which provides permission for communication with the Amazon Kinesis data stream.

IAM policy for an unencrypted stream from Kinesis Data Streams

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "ReadStream",
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStreamSummary",
                "kinesis:GetShardIterator",
                "kinesis:GetRecords",
                "kinesis:DescribeStream"
            ],
            "Resource": "arn:aws:kinesis:*:0123456789:stream/*"
        },
        {
            "Sid": "ListStream",
            "Effect": "Allow",
            "Action": [
                "kinesis:ListStreams",
                "kinesis:ListShards"
            ],
            "Resource": "*"
        }
    ]
}

IAM policy for an encrypted stream from Kinesis Data Streams

{
	"Version": "2012-10-17",
	"Statement": [{
			"Sid": "ReadStream",
			"Effect": "Allow",
			"Action": [
				"kinesis:DescribeStreamSummary",
				"kinesis:GetShardIterator",
				"kinesis:GetRecords",
				"kinesis:DescribeStream"
			],
			"Resource": "arn:aws:kinesis:*:0123456789:stream/*"
		},
		{
			"Sid": "DecryptStream",
			"Effect": "Allow",
			"Action": [
				"kms:Decrypt"
			],
			"Resource": "arn:aws:kms:us-east-1:0123456789:key/1234abcd-12ab-34cd-56ef-1234567890ab"
		}, 
        {
			"Sid": "ListStream",
			"Effect": "Allow",
			"Action": [
				"kinesis:ListStreams",
				"kinesis:ListShards"
			],
			"Resource": "*"
		}
	]
}

Check your VPC and verify that your Amazon Redshift cluster or Amazon Redshift Serverless has a route to get to the Kinesis Data Streams endpoints over the internet using a NAT gateway or internet gateway. If you want traffic between Redshift and Kinesis Data Streams to remain within the Amazon network, consider using a Kinesis Interface VPC Endpoint. For more information, see Using Amazon Kinesis Data Streams Kinesis Data Streams with Interface VPC Endpoints.

In Amazon Redshift, create an external schema to map the data from Kinesis to a schema.

CREATE EXTERNAL SCHEMA kds
FROM KINESIS
IAM_ROLE { default | 'iam-role-arn' };

Streaming ingestion for Kinesis Data Streams doesn't require an authentication type. It uses the IAM role defined in the CREATE EXTERNAL SCHEMA statement for making Kinesis Data Streams requests.

Optional: Use the REGION keyword to specify the region where the Amazon Kinesis Data Streams or Amazon MSK stream resides.

CREATE EXTERNAL SCHEMA kds
FROM KINESIS
REGION 'us-west-2'
IAM_ROLE { default | 'iam-role-arn' };

In this sample, the region specifies the location of the source stream. The IAM_ROLE is a sample.

Create a materialized view to consume the stream data. With a statement like the following, if a record can't be parsed, it causes an error. Use a command like this if you don't want error records to be skipped.

CREATE MATERIALIZED VIEW my_view AUTO REFRESH YES AS
SELECT *
FROM kds.my_stream_name;

The following example defines a materialized view for source data in JSON format. The view validates that incoming data is properly formatted JSON. Kinesis stream names are case sensitive and can contain both uppercase and lowercase letters. To ingest from streams with uppercase names, you can set the configuration enable_case_sensitive_identifier to true at the database level. For more information, see Names and identifiers and enable_case_sensitive_identifier.

CREATE MATERIALIZED VIEW my_view AUTO REFRESH YES AS
SELECT approximate_arrival_timestamp,
partition_key,
shard_id,
sequence_number,
refresh_time,
JSON_PARSE(kinesis_data) as kinesis_data
FROM kds.my_stream_name
WHERE CAN_JSON_PARSE(kinesis_data);

To turn on auto refresh, use AUTO REFRESH YES. The default behavior is manual refresh. Note when you use CAN_JSON_PARSE, it's possible that records that can't be parsed are skipped.

Metadata columns include the following:

It's important to note if you have business logic in your materialized view definition that business-logic errors can cause streaming ingestion to be blocked in some cases. This might lead to you having to drop and re-create the materialized view. To avoid this, we recommend that you keep your logic as simple as possible and perform most of your business-logic checks on the data after it's ingested.

Refresh the view, which invokes Redshift to read from the stream and load data into the materialized view.

REFRESH MATERIALIZED VIEW my_view;

Query data in the materialized view.

select * from my_view;