Managing Amazon Kinesis Data Streams

{
  "$schema": "http://json-schema.org/draft-04/schema#",
  "title": "Event",
  "type": "object",
  "additionalProperties": false,
  "properties": {
    "id":         {"type": "integer"},
    "event_time": {"type": "string", "format": "date-time"},
    "source":     {"type": "string"},
    "value":      {"type": "number"}
  },
  "required": ["id", "event_time", "value"]
}

aws glue create-registry --registry-name my-kds-registry --region $REGION

aws glue create-schema \
  --registry-id RegistryName=my-kds-registry \
  --schema-name my-event-schema \
  --data-format JSON \
  --compatibility NONE \
  --schema-definition '<schema JSON above>'

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": {"Service": "kinesis.amazonaws.com"},
    "Action": "sts:AssumeRole",
    "Condition": {
      "StringEquals": {"aws:SourceAccount": "ACCOUNT_ID"},
      "ArnLike": {"aws:SourceArn": "arn:aws:kinesis:REGION:ACCOUNT_ID:channel/*"}
    }
  }]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "S3TablesAccess",
      "Effect": "Allow",
      "Action": [
        "s3tables:GetTable",
        "s3tables:GetTableBucket",
        "s3tables:GetTableMetadataLocation",
        "s3tables:UpdateTableMetadataLocation",
        "s3tables:GetNamespace",
        "s3tables:CreateNamespace",
        "s3tables:CreateTable",
        "s3tables:PutTableData",
        "s3tables:GetTableData",
        "s3tables:PutTableEncryption",
        "s3tables:PutTableRecordExpirationConfiguration",
        "s3tables:TagResource"
      ],
      "Resource": [
        "arn:aws:s3tables:REGION:ACCOUNT_ID:bucket/TABLE_BUCKET",
        "arn:aws:s3tables:REGION:ACCOUNT_ID:bucket/TABLE_BUCKET/table/*"
      ]
    },
    {
      "Sid": "DLQBucketAccess",
      "Effect": "Allow",
      "Action": [
        "s3:PutObject",
        "s3:GetBucketLocation",
        "s3:ListBucket",
        "s3:ListBucketMultipartUploads",
        "s3:UploadPart",
        "s3:CompleteMultipartUpload",
        "s3:CreateMultipartUpload"
      ],
      "Resource": [
        "arn:aws:s3:::DLQ_BUCKET",
        "arn:aws:s3:::DLQ_BUCKET/*"
      ]
    },
    {
      "Sid": "GlueSchemaRegistryAccess",
      "Effect": "Allow",
      "Action": [
        "glue:GetRegistry",
        "glue:GetSchema",
        "glue:GetSchemaVersion",
        "glue:GetSchemaByDefinition",
        "glue:ListRegistries",
        "glue:ListSchemas"
      ],
      "Resource": [
        "arn:aws:glue:REGION:ACCOUNT_ID:registry/REGISTRY",
        "arn:aws:glue:REGION:ACCOUNT_ID:schema/REGISTRY/SCHEMA"
      ]
    },
    {
      "Sid": "KMSAccess",
      "Effect": "Allow",
      "Action": ["kms:Decrypt", "kms:GenerateDataKey", "kms:DescribeKey"],
      "Resource": "arn:aws:kms:REGION:ACCOUNT_ID:key/KEY_ID"
    },
    {
      "Sid": "KMSAccessDLQ",
      "Effect": "Allow",
      "Action": ["kms:Decrypt", "kms:GenerateDataKey"],
      "Resource": "arn:aws:kms:REGION:ACCOUNT_ID:key/DLQ_KEY_ID",
      "Condition": {
        "StringEquals": {"kms:ViaService": "s3.REGION.amazonaws.com"},
        "StringLike": {"kms:EncryptionContext:aws:s3:arn": "arn:aws:s3:::DLQ_BUCKET/*"}
      }
    },
    {
      "Sid": "CloudWatchLogsAccess",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents",
        "logs:DescribeLogStreams"
      ],
      "Resource": "arn:aws:logs:REGION:ACCOUNT_ID:log-group:/aws/kinesis/CHANNEL_NAME:*"
    }
  ]
}

{
  "Sid": "AllowS3TablesMaintenance",
  "Effect": "Allow",
  "Principal": {"Service": "maintenance.s3tables.amazonaws.com"},
  "Action": ["kms:Decrypt", "kms:GenerateDataKey", "kms:DescribeKey"],
  "Resource": "*",
  "Condition": {
    "StringEquals": {"aws:SourceAccount": "ACCOUNT_ID"},
    "ArnLike": {"aws:SourceArn": "arn:aws:s3tables:REGION:ACCOUNT_ID:bucket/TABLE_BUCKET*"}
  }
}

{
  "Version": "2012-10-17",
  "Statement": [{
    "Sid": "CrossAccountTableBucketAccess",
    "Effect": "Allow",
    "Principal": {"AWS": "arn:aws:iam::SOURCE_ACCOUNT_ID:role/EXECUTION_ROLE_NAME"},
    "Action": [
      "s3tables:GetTable",
      "s3tables:GetTableBucket",
      "s3tables:GetTableMetadataLocation",
      "s3tables:UpdateTableMetadataLocation",
      "s3tables:GetNamespace",
      "s3tables:CreateNamespace",
      "s3tables:CreateTable",
      "s3tables:PutTableData"
    ],
    "Resource": [
      "arn:aws:s3tables:REGION:DEST_ACCOUNT_ID:bucket/TABLE_BUCKET",
      "arn:aws:s3tables:REGION:DEST_ACCOUNT_ID:bucket/TABLE_BUCKET/table/*"
    ]
  }]
}

# S3 Tables bucket
aws s3tables create-table-bucket --name my-table-bucket --region $REGION

# DLQ bucket
aws s3 mb s3://my-kds-iceberg-dlq-$ACCOUNT --region $REGION

# CW log group — create before the channel
aws logs create-log-group \
  --log-group-name "/aws/kinesis/my-iceberg-channel" \
  --region $REGION
aws logs put-retention-policy \
  --log-group-name "/aws/kinesis/my-iceberg-channel" \
  --retention-in-days 30 \
  --region $REGION

cat > stream-config.json << 'EOF'
[{
  "StreamARN": "STREAM_ARN",
  "RecordConfiguration": {
    "RecordFormatType": "GSR_JSON",
    "GSRSchemaARN": "arn:aws:glue:REGION:ACCOUNT_ID:schema/my-kds-registry/my-event-schema"
  }
}]
EOF

cat > s3-tables-dest.json << 'EOF'
{
  "DataFreshnessInSeconds": 300,
  "DeadLetterQueueS3Configuration": {
    "BucketARN": "arn:aws:s3:::my-kds-iceberg-dlq-ACCOUNT_ID",
    "ExpectedBucketOwner": "ACCOUNT_ID",
    "ErrorOutputPrefix": "errors/my-iceberg-channel/"
  },
  "S3TablesConfigurationList": [{
    "TableBucketARN": "arn:aws:s3tables:REGION:ACCOUNT_ID:bucket/my-table-bucket",
    "Namespace": "my_namespace",
    "TableName": "my_table",
    "CompressionType": "ZSTD",
    "PartitionSpec": {
      "PartitionFields": [{"Transform": "TIME_HOUR", "SourceName": "event_time"}]
    }
  }]
}
EOF

cat > logging.json << 'EOF'
{"CloudWatchLogs": {"Enabled": true, "LogGroupName": "/aws/kinesis/my-iceberg-channel", "LogStreamName": "DestinationDelivery"}}
EOF

aws kinesis create-channel \
  --region $REGION \
  --channel-name my-iceberg-channel \
  --service-execution-role-arn $SER_ARN \
  --stream-configuration-list file://stream-config.json \
  --s3-tables-destination-configuration file://s3-tables-dest.json \
  --logging-configuration file://logging.json

aws kinesis describe-channel \
  --region $REGION \
  --channel-arn $CHANNEL_ARN \
  --query 'ChannelDescription.ChannelStatus'

from aws_schema_registry import SchemaRegistryClient, KafkaSerializer, DataAndSchema
from aws_schema_registry.jsonschema import JsonSchema
import boto3

session = boto3.Session(region_name=REGION)
registry_client = SchemaRegistryClient(session.client("glue"), registry_name="my-kds-registry")
serializer = KafkaSerializer(
    client=registry_client,
    schema_naming_strategy=lambda topic, is_key, schema: "my-event-schema",
)
json_schema = JsonSchema(SCHEMA_DEF)

record = {"id": 1, "event_time": "2026-07-09T20:00:00Z", "source": "app", "value": 1.5}
gsr_bytes = serializer.serialize(
    topic=STREAM_NAME,
    data_and_schema=DataAndSchema(data=record, schema=json_schema),
)
session.client("kinesis").put_record(
    StreamName=STREAM_NAME, Data=gsr_bytes, PartitionKey="1"
)

# Describe
aws kinesis describe-channel --region $REGION --channel-arn $CHANNEL_ARN

# List channels on a stream
aws kinesis list-channels --region $REGION --stream-filter "{\"StreamARN\": \"$STREAM_ARN\"}"

# Update — only DataFreshnessInSeconds and logging are mutable
aws kinesis update-channel --region $REGION \
  --channel-arn $CHANNEL_ARN \
  --s3-tables-destination-configuration '{"DataFreshnessInSeconds": 600}'

# Delete — already-delivered data is NOT removed
aws kinesis delete-channel --region $REGION \
  --channel-arn $CHANNEL_ARN

# Namespace appears within ~1 min of the first delivery attempt
aws s3tables list-namespaces --table-bucket-arn $TABLE_BUCKET_ARN --region $REGION

# Table appears after the first successful delivery cycle
aws s3tables list-tables --table-bucket-arn $TABLE_BUCKET_ARN \
  --namespace my_namespace --region $REGION

# Failed records
aws s3 ls s3://my-kds-iceberg-dlq-$ACCOUNT/errors/ --recursive

# Delivery errors
aws logs filter-log-events \
  --log-group-name "/aws/kinesis/my-iceberg-channel" --region $REGION

aws lakeformation grant-permissions \
  --principal DataLakePrincipalIdentifier="arn:aws:iam::ACCOUNT_ID:role/QUERY_ROLE" \
  --resource '{"Database":{"CatalogId":"ACCOUNT_ID:s3tablescatalog/TABLE_BUCKET","Name":"my_namespace"}}' \
  --permissions DESCRIBE SELECT

aws lakeformation grant-permissions \
  --principal DataLakePrincipalIdentifier="arn:aws:iam::ACCOUNT_ID:role/QUERY_ROLE" \
  --resource '{"Table":{"CatalogId":"ACCOUNT_ID:s3tablescatalog/TABLE_BUCKET","DatabaseName":"my_namespace","Name":"my_table"}}' \
  --permissions SELECT DESCRIBE

SELECT * FROM "s3tablescatalog/TABLE_BUCKET"."my_namespace"."my_table" LIMIT 10;

aws s3tables put-table-record-expiration-configuration ...