AWS CloudFormation
ユーザーガイド (API バージョン 2010-05-15)

AWS::KinesisFirehose::DeliveryStream

AWS::KinesisFirehose::DeliveryStream リソースは、Amazon Simple Storage Service (Amazon S3)、Amazon Redshift、または Amazon Elasticsearch Service (Amazon ES) 送信先にリアルタイムのストリーミングデータを送信する、Amazon Kinesis Data Firehose (Kinesis Data Firehose) 配信ストリームを作成します。詳細については、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 の送信先。

条件付き。送信先の設定は 1 つだけ指定する必要があります。

Amazon ES の送信先から Amazon S3 または Amazon Redshift の送信先に送信ストリームの送信先を変更する場合、更新するには一時的な中断が必要です。

必須: 条件付き

タイプ: ElasticsearchDestinationConfiguration

Update requires: No interruption

ExtendedS3DestinationConfiguration

送信ストリームの Amazon S3 の送信先。

条件付き。送信先の設定は 1 つだけ指定する必要があります。

Amazon Extended S3 の送信先から Amazon ES の送信先に送信ストリームの送信先を変更する場合、更新するには一時的な中断が必要です。

必須: 条件付き

タイプ: ExtendedS3DestinationConfiguration

Update requires: No interruption

KinesisStreamSourceConfiguration

Kinesis ストリームを配信ストリームのソースとして使用する場合、ソースストリームの Kinesis Stream ARN とロール ARN を含む KinesisStreamSourceConfiguration

必須: いいえ

タイプ: KinesisStreamSourceConfiguration

Update requires: No interruption

RedshiftDestinationConfiguration

配信ストリームの Amazon Redshift の送信先。

条件付き。送信先の設定は 1 つだけ指定する必要があります。

Amazon Redshift の送信先から Amazon ES の送信先に送信ストリームの送信先を変更する場合、更新するには一時的な中断が必要です。

必須: 条件付き

タイプ: RedshiftDestinationConfiguration

Update requires: No interruption

S3DestinationConfiguration

S3DestinationConfiguration プロパティタイプは、Amazon Kinesis Data Firehose (Kinesis Data Firehose) がデータを配信する Amazon Simple Storage Service (Amazon S3) 送信先を指定します。

条件付き。送信先の設定は 1 つだけ指定する必要があります。

Amazon S3 の送信先から Amazon ES の送信先に送信ストリームの送信先を変更する場合、更新するには一時的な中断が必要です。

必須: 条件付き

タイプ: S3DestinationConfiguration

Update requires: No interruption

SplunkDestinationConfiguration

Splunk の配信ストリームの送信先の設定。

必須: いいえ

タイプ: SplunkDestinationConfiguration

Update requires: No interruption

戻り値

参照番号

このリソースの論理 ID が Ref 組み込み関数に提供されると、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 Stream を指定する

以下の例では、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

以下の資料も参照してください。