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.

Arn: The Amazon Resource Name (ARN) of the delivery stream, such as arn:aws:firehose:us-east-2:123456789012:deliverystream/delivery-stream-name.

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: {}
            #Name: kinesislab2
            
            GlueTable:
            Type: AWS::Glue::Table
            Properties:
            CatalogId: !Ref AWS::AccountId
            DatabaseName: !Ref GlueDatabase
            TableInput:
            #Name: nyc-taxi-trips2
            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
            #Location: !Join 
            #- ''
            #- - 's3://'
            #  - !Ref s3bucket
            #  - 'nyc-taxi-trips2'
            #  - '/'
            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: 
            #DeliveryStreamName: nyc-taxi-trips2
            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

AWS::KinesisFirehose::DeliveryStream

Syntax

JSON

YAML

Properties

Return Values

Ref

Fn::GetAtt

Examples

Create a Kinesis Data Firehose Delivery Stream

JSON

YAML

Convert Record Format

YAML

Specify an Amazon S3 Destination for the Delivery Stream

JSON

YAML

Specify a Kinesis Stream as the Source for the Delivery Stream

JSON

YAML

See Also