Creates a channel that delivers records from a Kinesis data stream to a destination¶

Description¶

Creates a channel that delivers records from a Kinesis data stream to a destination. A channel reads records from the specified stream and writes them to streaming tables on Apache Iceberg (Amazon S3 Tables) or to a general purpose Amazon S3 bucket.

You must specify either S3DestinationConfiguration or S3TablesDestinationConfiguration, but not both.

To use this operation, you must have permission to pass the specified service execution IAM role to Amazon Kinesis Data Streams (the iam:PassRole permission on that role).

Creating a channel is an asynchronous operation. Upon receiving the request, Amazon Kinesis Data Streams returns immediately with the channel in the CREATING state. After provisioning is complete, Amazon Kinesis Data Streams sets the state to ACTIVE. You can use describe_channel to check the current state.

This operation is only supported for data streams with the on-demand capacity mode.

This operation has a call limit of 5 transactions per second (TPS) for each Amazon Web Services account. Exceeding 5 TPS results in a LimitExceededException.

Usage¶

kinesis_create_channel(ChannelName, ServiceExecutionRoleARN,
  StreamConfigurationList, S3DestinationConfiguration,
  S3TablesDestinationConfiguration, EncryptionConfiguration, Tags,
  LoggingConfiguration)

Arguments¶

ChannelName

[required] The name of the channel. The name is unique within your Amazon Web Services account and Amazon Web Services Region.

ServiceExecutionRoleARN

[required] The Amazon Resource Name (ARN) of the IAM role that Amazon Kinesis Data Streams assumes to write records to the destination.

StreamConfigurationList

[required] The source stream configuration for the channel. Currently, one stream is supported per channel.

S3DestinationConfiguration

The configuration for delivery to a general purpose Amazon S3 bucket. Specify this parameter when S3TablesDestinationConfiguration is not specified.

S3TablesDestinationConfiguration

The configuration for delivery to streaming tables on Apache Iceberg in Amazon S3 Tables. Specify this parameter when S3DestinationConfiguration is not specified.

EncryptionConfiguration

The server-side encryption configuration that uses an Amazon Web Services KMS key to encrypt data delivered to the destination.

Tags

A set of key-value pairs to assign to the channel. A tag consists of a required key and an optional value.

LoggingConfiguration

The Amazon CloudWatch Logs configuration for the channel.

Value¶

A list with the following syntax:

list(
  ChannelDescription = list(
    ChannelName = "string",
    ChannelARN = "string",
    ChannelId = "string",
    ChannelStatus = "CREATING"|"ACTIVE"|"UPDATING"|"DELETING"|"FAILED",
    ChannelStatusReason = "string",
    ChannelCreationTimestamp = as.POSIXct(
      "2015-01-01"
    ),
    ServiceExecutionRoleARN = "string",
    StreamConfigurationList = list(
      list(
        StreamARN = "string",
        StreamCreationTimestamp = as.POSIXct(
          "2015-01-01"
        ),
        RecordConfiguration = list(
          RecordFormatType = "GSR_JSON"|"JSON"|"STRING"|"BYTE_ARRAY",
          GSRSchemaARN = "string"
        )
      )
    ),
    S3DestinationConfiguration = list(
      DataFreshnessInSeconds = 123,
      DeadLetterQueueS3Configuration = list(
        BucketARN = "string",
        ExpectedBucketOwner = "string",
        ErrorOutputPrefix = "string"
      ),
      StorageConfiguration = list(
        BucketARN = "string",
        ExpectedBucketOwner = "string",
        OutputKeyTemplate = "string",
        StorageClass = "STANDARD"|"INTELLIGENT_TIERING"|"GLACIER_IR",
        CompressionType = "NONE"|"GZIP"|"ZSTD"
      )
    ),
    S3TablesDestinationConfiguration = list(
      DataFreshnessInSeconds = 123,
      DeadLetterQueueS3Configuration = list(
        BucketARN = "string",
        ExpectedBucketOwner = "string",
        ErrorOutputPrefix = "string"
      ),
      S3TablesConfigurationList = list(
        list(
          TableBucketARN = "string",
          Namespace = "string",
          TableName = "string",
          CompressionType = "NONE"|"ZSTD"|"SNAPPY",
          PartitionSpec = list(
            PartitionFields = list(
              list(
                Transform = "TIME_HOUR",
                SourceName = "string"
              )
            )
          )
        )
      )
    ),
    EncryptionConfiguration = list(
      EncryptionType = "KMS",
      KeyId = "string"
    ),
    LoggingConfiguration = list(
      CloudWatchLogs = list(
        Enabled = TRUE|FALSE,
        LogGroupName = "string",
        LogStreamName = "string"
      )
    )
  )
)

Request syntax¶

svc$create_channel(
  ChannelName = "string",
  ServiceExecutionRoleARN = "string",
  StreamConfigurationList = list(
    list(
      StreamARN = "string",
      RecordConfiguration = list(
        RecordFormatType = "GSR_JSON"|"JSON"|"STRING"|"BYTE_ARRAY",
        GSRSchemaARN = "string"
      )
    )
  ),
  S3DestinationConfiguration = list(
    DataFreshnessInSeconds = 123,
    DeadLetterQueueS3Configuration = list(
      BucketARN = "string",
      ExpectedBucketOwner = "string",
      ErrorOutputPrefix = "string"
    ),
    StorageConfiguration = list(
      BucketARN = "string",
      ExpectedBucketOwner = "string",
      OutputKeyTemplate = "string",
      StorageClass = "STANDARD"|"INTELLIGENT_TIERING"|"GLACIER_IR",
      CompressionType = "NONE"|"GZIP"|"ZSTD"
    )
  ),
  S3TablesDestinationConfiguration = list(
    DataFreshnessInSeconds = 123,
    DeadLetterQueueS3Configuration = list(
      BucketARN = "string",
      ExpectedBucketOwner = "string",
      ErrorOutputPrefix = "string"
    ),
    S3TablesConfigurationList = list(
      list(
        TableBucketARN = "string",
        Namespace = "string",
        TableName = "string",
        CompressionType = "NONE"|"ZSTD"|"SNAPPY",
        PartitionSpec = list(
          PartitionFields = list(
            list(
              Transform = "TIME_HOUR",
              SourceName = "string"
            )
          )
        )
      )
    )
  ),
  EncryptionConfiguration = list(
    EncryptionType = "KMS",
    KeyId = "string"
  ),
  Tags = list(
    "string"
  ),
  LoggingConfiguration = list(
    CloudWatchLogs = list(
      Enabled = TRUE|FALSE,
      LogGroupName = "string",
      LogStreamName = "string"
    )
  )
)