AWS::KinesisFirehose::DeliveryStream - AWS CloudFormation
AWS 文档中描述的 AWS 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 AWS 服务入门

AWS::KinesisFirehose::DeliveryStream

AWS::KinesisFirehose::DeliveryStream 资源创建一个 Amazon Kinesis Data Firehose (Kinesis Data Firehose) 传输流,它将实时流数据传输到 Amazon Simple Storage Service (Amazon S3)、Amazon Redshift 或 Amazon Elasticsearch Service (Amazon ES) 目标。有关更多信息,请参阅 Amazon Kinesis Data Firehose 开发人员指南 中的创建 Amazon Kinesis Data Firehose 传输流

语法

要在 AWS CloudFormation 模板中声明此实体,请使用以下语法:

属性

DeliveryStreamName

传输流名称。

必需:否

类型:字符串

最低1

最高64

模式[a-zA-Z0-9_.-]+

Update requires: Replacement

DeliveryStreamType

传输流类型。它可以是以下值之一:

  • DirectPut:提供商应用程序直接访问传输流。

  • KinesisStreamAsSource:传输流使用 Kinesis 数据流作为源。

必需:否

类型:字符串

允许的值DirectPut | KinesisStreamAsSource

Update requires: Replacement

ElasticsearchDestinationConfiguration

传输流的 Amazon ES 目标。

这是有条件的。您必须仅指定一个目标配置。

如果您将传输流目标从 Amazon ES 目标更改为 Amazon S3 或 Amazon Redshift 目标,则更新需要一些中断

必需:条件

类型ElasticsearchDestinationConfiguration

Update requires: No interruption

ExtendedS3DestinationConfiguration

传输流的 Amazon S3 目标。

这是有条件的。您必须仅指定一个目标配置。

如果您将传输流目标从 Amazon Extended S3 目标更改为 Amazon ES 目标,则更新需要一些中断

必需:条件

类型ExtendedS3DestinationConfiguration

Update requires: No interruption

HttpEndpointDestinationConfiguration

允许配置 Kinesis Firehose 以将数据传送到任何 HTTP 终端节点目标。您只能指定一个目标。

必需:否

类型HttpEndpointDestinationConfiguration

Update requires: No interruption

KinesisStreamSourceConfiguration

当 Kinesis 流用作传输流的源时,KinesisStreamSourceConfiguration 包含 Kinesis 流 ARN 和源流的角色 ARN。

必需:否

类型KinesisStreamSourceConfiguration

Update requires: No interruption

RedshiftDestinationConfiguration

传输流的 Amazon Redshift 目标。

这是有条件的。您必须仅指定一个目标配置。

如果您将传输流目标从 Amazon Redshift 目标更改为 Amazon ES 目标,则更新需要一些中断

必需:条件

类型RedshiftDestinationConfiguration

Update requires: No interruption

S3DestinationConfiguration

S3DestinationConfiguration 属性类型指定 Amazon Kinesis Data Firehose (Kinesis Data Firehose) 将数据传输到的 Amazon Simple Storage Service (Amazon S3) 目标。

这是有条件的。您必须仅指定一个目标配置。

如果您将传输流目标从 Amazon S3 目标更改为 Amazon ES 目标,则更新需要一些中断

必需:条件

类型S3DestinationConfiguration

Update requires: No interruption

SplunkDestinationConfiguration

传输流的 Splunk 目标配置。

必需:否

类型SplunkDestinationConfiguration

Update requires: No interruption

返回值

Ref

当向 Ref 内部函数提供此资源的逻辑 ID 时,Ref 将返回传输流名称,例如 mystack-deliverystream-1ABCD2EF3GHIJ

有关使用 Ref 函数的更多信息,请参阅 Ref

Fn::GetAtt

Fn::GetAtt 返回此类型的指定属性的值。以下为可用属性和示例返回值。

有关使用 Fn::GetAtt 的更多信息,请参阅 Fn::GetAtt

Arn

传输流的 Amazon 资源名称 (ARN),例如 arn:aws:firehose:us-east-2:123456789012:deliverystream/delivery-stream-name

示例

创建 Kinesis Data Firehose 传输流

以下示例创建一个将数据传输到 Amazon ES 目标的 Kinesis Data Firehose 传输流。Kinesis Data Firehose 将在 Amazon S3 存储桶中备份已发送到目标的所有数据。

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"

转换记录格式

以下示例显示记录格式转换。

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

指定传输流的 Amazon S3 目标

以下示例使用 ExtendedS3DestinationConfiguration 属性指定传输流的 Amazon S3 目标。

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

指定 Kinesis 流作为传输流的源

以下示例使用 KinesisStreamSourceConfiguration 属性指定 Kinesis 流作为传输流的来源。

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

另请参阅