AWS::KinesisFirehose::DeliveryStream
The AWS::KinesisFirehose::DeliveryStream
resource creates an Amazon
Kinesis Data Firehose (Kinesis Data Firehose) delivery stream that delivers real-time
streaming data to an Amazon Simple Storage Service (Amazon S3), Amazon Redshift, or
Amazon
Elasticsearch Service (Amazon ES) destination. For more information, see Creating an Amazon
Kinesis Data Firehose Delivery Stream in the Amazon Kinesis Data
Firehose Developer Guide.
Syntax
To declare this entity in your AWS CloudFormation template, use the following syntax:
JSON
{ "Type" : "AWS::KinesisFirehose::DeliveryStream", "Properties" : { "DeliveryStreamName" :
String
, "DeliveryStreamType" :String
, "ElasticsearchDestinationConfiguration" :ElasticsearchDestinationConfiguration
, "ExtendedS3DestinationConfiguration" :ExtendedS3DestinationConfiguration
, "KinesisStreamSourceConfiguration" :KinesisStreamSourceConfiguration
, "RedshiftDestinationConfiguration" :RedshiftDestinationConfiguration
, "S3DestinationConfiguration" :S3DestinationConfiguration
, "SplunkDestinationConfiguration" :SplunkDestinationConfiguration
} }
YAML
Type: AWS::KinesisFirehose::DeliveryStream Properties: DeliveryStreamName:
String
DeliveryStreamType:String
ElasticsearchDestinationConfiguration:ElasticsearchDestinationConfiguration
ExtendedS3DestinationConfiguration:ExtendedS3DestinationConfiguration
KinesisStreamSourceConfiguration:KinesisStreamSourceConfiguration
RedshiftDestinationConfiguration:RedshiftDestinationConfiguration
S3DestinationConfiguration:S3DestinationConfiguration
SplunkDestinationConfiguration:SplunkDestinationConfiguration
Properties
DeliveryStreamName
-
The name of the delivery stream.
Required: No
Type: String
Minimum:
1
Maximum:
64
Pattern:
[a-zA-Z0-9_.-]+
Update requires: Replacement
DeliveryStreamType
-
The delivery stream type. This can be one of the following values:
-
DirectPut
: Provider applications access the delivery stream directly. -
KinesisStreamAsSource
: The delivery stream uses a Kinesis data stream as a source.
Required: No
Type: String
Allowed Values:
DirectPut | KinesisStreamAsSource
Update requires: Replacement
-
ElasticsearchDestinationConfiguration
-
An Amazon ES destination for the delivery stream.
Conditional. You must specify only one destination configuration.
If you change the delivery stream destination from an Amazon ES destination to an Amazon S3 or Amazon Redshift destination, update requires some interruptions.
Required: Conditional
Type: ElasticsearchDestinationConfiguration
Update requires: No interruption
ExtendedS3DestinationConfiguration
-
An Amazon S3 destination for the delivery stream.
Conditional. You must specify only one destination configuration.
If you change the delivery stream destination from an Amazon Extended S3 destination to an Amazon ES destination, update requires some interruptions.
Required: Conditional
Type: ExtendedS3DestinationConfiguration
Update requires: No interruption
KinesisStreamSourceConfiguration
-
When a Kinesis stream is used as the source for the delivery stream, a KinesisStreamSourceConfiguration containing the Kinesis stream ARN and the role ARN for the source stream.
Required: No
Type: KinesisStreamSourceConfiguration
Update requires: No interruption
RedshiftDestinationConfiguration
-
An Amazon Redshift destination for the delivery stream.
Conditional. You must specify only one destination configuration.
If you change the delivery stream destination from an Amazon Redshift destination to an Amazon ES destination, update requires some interruptions.
Required: Conditional
Type: RedshiftDestinationConfiguration
Update requires: No interruption
S3DestinationConfiguration
-
The
S3DestinationConfiguration
property type specifies an Amazon Simple Storage Service (Amazon S3) destination to which Amazon Kinesis Data Firehose (Kinesis Data Firehose) delivers data.Conditional. You must specify only one destination configuration.
If you change the delivery stream destination from an Amazon S3 destination to an Amazon ES destination, update requires some interruptions.
Required: Conditional
Type: S3DestinationConfiguration
Update requires: No interruption
SplunkDestinationConfiguration
-
The configuration of a destination in Splunk for the delivery stream.
Required: No
Type: SplunkDestinationConfiguration
Update requires: No interruption
Return Values
Ref
When the logical ID of this resource is provided to the Ref intrinsic function, Ref
returns the delivery stream name, such as
mystack-deliverystream-1ABCD2EF3GHIJ
.
For more information about using the Ref function, see Ref.
Fn::GetAtt
Fn::GetAtt returns a value for a specified attribute of this type. The following are the available attributes and sample return values.
For more information about using Fn::GetAtt, see Fn::GetAtt.
Examples
Create a Kinesis Data Firehose Delivery Stream
The following example creates a Kinesis Data Firehose delivery stream that delivers data to an Amazon ES destination. Kinesis Data Firehose backs up all data sent to the destination in an Amazon S3 bucket.
JSON
"ElasticSearchDeliveryStream": { "Type": "AWS::KinesisFirehose::DeliveryStream", "Properties": { "ElasticsearchDestinationConfiguration": { "BufferingHints": { "IntervalInSeconds": 60, "SizeInMBs": 50 }, "CloudWatchLoggingOptions": { "Enabled": true, "LogGroupName": "deliverystream", "LogStreamName": "elasticsearchDelivery" }, "DomainARN": { "Ref" : "MyDomainARN" }, "IndexName": { "Ref" : "MyIndexName" }, "IndexRotationPeriod": "NoRotation", "TypeName" : "fromFirehose", "RetryOptions": { "DurationInSeconds": "60" }, "RoleARN": { "Fn::GetAtt" : ["ESdeliveryRole", "Arn"] }, "S3BackupMode": "AllDocuments", "S3Configuration": { "BucketARN": { "Ref" : "MyBackupBucketARN" }, "BufferingHints": { "IntervalInSeconds": "60", "SizeInMBs": "50" }, "CompressionFormat": "UNCOMPRESSED", "Prefix": "firehose/", "RoleARN": { "Fn::GetAtt" : ["S3deliveryRole", "Arn"] }, "CloudWatchLoggingOptions" : { "Enabled" : true, "LogGroupName" : "deliverystream", "LogStreamName" : "s3Backup" } } } } }
YAML
ElasticSearchDeliveryStream: Type: AWS::KinesisFirehose::DeliveryStream Properties: ElasticsearchDestinationConfiguration: BufferingHints: IntervalInSeconds: 60 SizeInMBs: 50 CloudWatchLoggingOptions: Enabled: true LogGroupName: "deliverystream" LogStreamName: "elasticsearchDelivery" DomainARN: Ref: "MyDomainARN" IndexName: Ref: "MyIndexName" IndexRotationPeriod: "NoRotation" TypeName: "fromFirehose" RetryOptions: DurationInSeconds: "60" RoleARN: Fn::GetAtt: - "ESdeliveryRole" - "Arn" S3BackupMode: "AllDocuments" S3Configuration: BucketARN: Ref: "MyBackupBucketARN" BufferingHints: IntervalInSeconds: "60" SizeInMBs: "50" CompressionFormat: "UNCOMPRESSED" Prefix: "firehose/" RoleARN: Fn::GetAtt: - "S3deliveryRole" - "Arn" CloudWatchLoggingOptions: Enabled: true LogGroupName: "deliverystream" LogStreamName: "s3Backup"
Convert Record Format
The following example shows record format conversion.
YAML
AWSTemplateFormatVersion: 2010-09-09 Description: Stack for Firehose DeliveryStream S3 Destination. Resources: GlueDatabase: Type: AWS::Glue::Database Properties: CatalogId: !Ref AWS::AccountId DatabaseInput: {} GlueTable: Type: AWS::Glue::Table Properties: CatalogId: !Ref AWS::AccountId DatabaseName: !Ref GlueDatabase TableInput: Owner: owner Retention: 0 StorageDescriptor: Columns: - Name: pickup_latitude Type: double - Name: pickup_longitude Type: double - Name: dropoff_latitude Type: double - Name: dropoff_longitude Type: double - Name: trip_id Type: int - Name: trip_distance Type: double - Name: passenger_count Type: int - Name: pickup_datetime Type: timestamp - Name: dropoff_datetime Type: timestamp - Name: total_amount Type: double InputFormat: org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat OutputFormat: org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat Compressed: false NumberOfBuckets: -1 SerdeInfo: SerializationLibrary: org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe Parameters: serialization.format: '1' BucketColumns: [] SortColumns: [] StoredAsSubDirectories: false PartitionKeys: - Name: year Type: string - Name: month Type: string - Name: day Type: string - Name: hour Type: string TableType: EXTERNAL_TABLE deliverystream: Type: AWS::KinesisFirehose::DeliveryStream Properties: DeliveryStreamType: DirectPut ExtendedS3DestinationConfiguration: RoleARN: !GetAtt deliveryRole.Arn BucketARN: !Join - '' - - 'arn:aws:s3:::' - !Ref s3bucket Prefix: !Join - '' - - !Ref GlueTable - '/year=!{timestamp:YYYY}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/' ErrorOutputPrefix: !Join - '' - - !Ref GlueTable - 'error/!{firehose:error-output-type}/year=!{timestamp:YYYY}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/' BufferingHints: SizeInMBs: 128 IntervalInSeconds: 300 CompressionFormat: UNCOMPRESSED EncryptionConfiguration: NoEncryptionConfig: NoEncryption CloudWatchLoggingOptions: Enabled: true LogGroupName: !Join - '' - - 'KDF-' - !Ref GlueTable LogStreamName: S3Delivery S3BackupMode: Disabled DataFormatConversionConfiguration: SchemaConfiguration: CatalogId: !Ref AWS::AccountId RoleARN: !GetAtt deliveryRole.Arn DatabaseName: !Ref GlueDatabase TableName: !Ref GlueTable Region: !Ref AWS::Region VersionId: LATEST InputFormatConfiguration: Deserializer: OpenXJsonSerDe: {} OutputFormatConfiguration: Serializer: ParquetSerDe: {} Enabled: True s3bucket: Type: AWS::S3::Bucket Properties: VersioningConfiguration: Status: Enabled deliveryRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Version: 2012-10-17 Statement: - Sid: '' Effect: Allow Principal: Service: firehose.amazonaws.com Action: 'sts:AssumeRole' Condition: StringEquals: 'sts:ExternalId': !Ref 'AWS::AccountId' Path: "/" Policies: - PolicyName: firehose_delivery_policy PolicyDocument: Version: 2012-10-17 Statement: - Effect: Allow Action: - 's3:AbortMultipartUpload' - 's3:GetBucketLocation' - 's3:GetObject' - 's3:ListBucket' - 's3:ListBucketMultipartUploads' - 's3:PutObject' Resource: - !Join - '' - - 'arn:aws:s3:::' - !Ref s3bucket - !Join - '' - - 'arn:aws:s3:::' - !Ref s3bucket - '*' - Effect: Allow Action: 'glue:GetTableVersions' Resource: '*' - Effect: Allow Action: 'logs:PutLogEvents' Resource: - !Join - '' - - 'arn:aws:logs:' - !Ref 'AWS::Region' - ':' - !Ref 'AWS::AccountId' - 'log-group:/aws/kinesisfirehose/KDF-' - !Ref GlueTable - ':log-stream:*' Outputs: deliverysreamARN: Description: The ARN of the firehose delivery stream Value: !GetAtt deliverystream.Arn
Specify an Amazon S3 Destination for the Delivery Stream
The following example uses the ExtendedS3DestinationConfiguration
property to specify an Amazon S3 destination for the delivery stream.
JSON
{ "AWSTemplateFormatVersion" : "2010-09-09", "Description" : "Stack for Firehose DeliveryStream S3 Destination.", "Resources": { "deliverystream": { "DependsOn": ["deliveryPolicy"], "Type": "AWS::KinesisFirehose::DeliveryStream", "Properties": { "ExtendedS3DestinationConfiguration": { "BucketARN": {"Fn::Join": ["", ["arn:aws:s3:::", {"Ref":"s3bucket"}]]}, "BufferingHints": { "IntervalInSeconds": "60", "SizeInMBs": "50" }, "CompressionFormat": "UNCOMPRESSED", "Prefix": "firehose/", "RoleARN": {"Fn::GetAtt" : ["deliveryRole", "Arn"] }, "ProcessingConfiguration" : { "Enabled": "true", "Processors": [ { "Parameters": [ { "ParameterName": "LambdaArn", "ParameterValue": {"Fn::GetAtt" : ["myLambda", "Arn"] } }], "Type": "Lambda" }] } } } }, "s3bucket": { "Type": "AWS::S3::Bucket", "Properties": { "VersioningConfiguration": { "Status": "Enabled" } } }, "deliveryRole": { "Type": "AWS::IAM::Role", "Properties": { "AssumeRolePolicyDocument": { "Version": "2012-10-17", "Statement": [ { "Sid": "", "Effect": "Allow", "Principal": { "Service": "firehose.amazonaws.com" }, "Action": "sts:AssumeRole", "Condition": { "StringEquals": { "sts:ExternalId": {"Ref":"AWS::AccountId"} } } } ] } } }, "deliveryPolicy": { "Type": "AWS::IAM::Policy", "Properties": { "PolicyName": "firehose_delivery_policy", "PolicyDocument": { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:AbortMultipartUpload", "s3:GetBucketLocation", "s3:GetObject", "s3:ListBucket", "s3:ListBucketMultipartUploads", "s3:PutObject" ], "Resource": [ {"Fn::Join": ["", ["arn:aws:s3:::", {"Ref":"s3bucket"}]]}, {"Fn::Join": ["", ["arn:aws:s3:::", {"Ref":"s3bucket"}, "*"]]} ] } ] }, "Roles": [{"Ref": "deliveryRole"}] } } } }
YAML
AWSTemplateFormatVersion: 2010-09-09 Description: Stack for Firehose DeliveryStream S3 Destination. Resources: deliverystream: DependsOn: - deliveryPolicy Type: AWS::KinesisFirehose::DeliveryStream Properties: ExtendedS3DestinationConfiguration: BucketARN: !Join - '' - - 'arn:aws:s3:::' - !Ref s3bucket BufferingHints: IntervalInSeconds: '60' SizeInMBs: '50' CompressionFormat: UNCOMPRESSED Prefix: firehose/ RoleARN: !GetAtt deliveryRole.Arn ProcessingConfiguration: Enabled: 'true' Processors: - Parameters: - ParameterName: LambdaArn ParameterValue: !GetAtt myLambda.Arn Type: Lambda s3bucket: Type: AWS::S3::Bucket Properties: VersioningConfiguration: Status: Enabled deliveryRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Version: 2012-10-17 Statement: - Sid: '' Effect: Allow Principal: Service: firehose.amazonaws.com Action: 'sts:AssumeRole' Condition: StringEquals: 'sts:ExternalId': !Ref 'AWS::AccountId' deliveryPolicy: Type: AWS::IAM::Policy Properties: PolicyName: firehose_delivery_policy PolicyDocument: Version: 2012-10-17 Statement: - Effect: Allow Action: - 's3:AbortMultipartUpload' - 's3:GetBucketLocation' - 's3:GetObject' - 's3:ListBucket' - 's3:ListBucketMultipartUploads' - 's3:PutObject' Resource: - !Join - '' - - 'arn:aws:s3:::' - !Ref s3bucket - !Join - '' - - 'arn:aws:s3:::' - !Ref s3bucket - '*' Roles: - !Ref deliveryRole
Specify a Kinesis Stream as the Source for the Delivery Stream
The following example uses the KinesisStreamSourceConfiguration
property to specify a Kinesis stream as the source for the delivery stream.
JSON
{ "Parameters": { "deliveryRoleArn": { "Type": "String" }, "deliveryStreamName": { "Type": "String" }, "kinesisStreamARN": { "Type": "String" }, "kinesisStreamRoleArn": { "Type": "String" }, "s3bucketArn": { "Type": "String" } }, "Resources": { "Deliverystream": { "Type": "AWS::KinesisFirehose::DeliveryStream", "Properties": { "DeliveryStreamName": { "Ref": "deliveryStreamName" }, "DeliveryStreamType": "KinesisStreamAsSource", "KinesisStreamSourceConfiguration": { "KinesisStreamARN": { "Ref": "kinesisStreamARN" }, "RoleARN": { "Ref": "kinesisStreamRoleArn" } }, "ExtendedS3DestinationConfiguration": { "BucketARN": { "Ref": "s3bucketArn" }, "BufferingHints": { "IntervalInSeconds": 60, "SizeInMBs": 50 }, "CompressionFormat": "UNCOMPRESSED", "Prefix": "firehose/", "RoleARN": { "Ref": "deliveryRoleArn" } } } } } }
YAML
Parameters: deliveryRoleArn: Type: String deliveryStreamName: Type: String kinesisStreamARN : Type : String kinesisStreamRoleArn: Type : String s3bucketArn: Type: String Resources : Deliverystream: Type: AWS::KinesisFirehose::DeliveryStream Properties: DeliveryStreamName: !Ref deliveryStreamName DeliveryStreamType: KinesisStreamAsSource KinesisStreamSourceConfiguration: KinesisStreamARN: !Ref kinesisStreamARN RoleARN: !Ref kinesisStreamRoleArn ExtendedS3DestinationConfiguration: BucketARN: !Ref s3bucketArn BufferingHints: IntervalInSeconds: 60 SizeInMBs: 50 CompressionFormat: UNCOMPRESSED Prefix: firehose/ RoleARN: !Ref deliveryRoleArn
See Also
-
CreateDeliveryStream in the Amazon Kinesis Data Firehose API Reference.