Amazon Data Firehose Construct Library
Amazon Data Firehose, formerly known as Amazon Kinesis Data Firehose, is a service for fully-managed delivery of real-time streaming data to storage services such as Amazon S3, Amazon Redshift, Amazon Elasticsearch, Splunk, or any custom HTTP endpoint or third-party services such as Datadog, Dynatrace, LogicMonitor, MongoDB, New Relic, and Sumo Logic.
Amazon Data Firehose delivery streams are distinguished from Kinesis data streams in their models of consumption. Whereas consumers read from a data stream by actively pulling data from the stream, a delivery stream pushes data to its destination on a regular cadence. This means that data streams are intended to have consumers that do on-demand processing, like AWS Lambda or Amazon EC2. On the other hand, delivery streams are intended to have destinations that are sources for offline processing and analytics, such as Amazon S3 and Amazon Redshift.
This module is part of the AWS Cloud Development Kit project. It allows you to define Amazon Data Firehose delivery streams.
Defining a Delivery Stream
In order to define a Delivery Stream, you must specify a destination. S3, HTTP endpoint, and Datadog can be used as destinations, as covered below.
ts
const bucket = new s3.Bucket(this, 'Bucket');
new firehose.DeliveryStream(this, 'Delivery Stream', {
destination: new firehose.S3Bucket(bucket),
});
python
bucket = s3.Bucket(self, "Bucket")
firehose.DeliveryStream(self, "Delivery Stream",
destination=firehose.S3Bucket(bucket)
)
java
Bucket bucket = new Bucket(this, "Bucket");
DeliveryStream.Builder.create(this, "Delivery Stream")
.destination(new S3Bucket(bucket))
.build();
csharp
var bucket = new Bucket(this, "Bucket");
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Destination = new S3Bucket(bucket)
});
go
bucket := s3.NewBucket(this, jsii.String("Bucket"))
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Destination: firehose.NewS3Bucket(bucket),
})
The above example defines the following resources:
- An S3 bucket
- An Amazon Data Firehose delivery stream with Direct PUT as the source and CloudWatch error logging turned on.
- An IAM role which gives the delivery stream permission to write to the S3 bucket.
Sources
An Amazon Data Firehose delivery stream can accept data from three main sources: Kinesis Data Streams, Managed Streaming for Apache Kafka (MSK), or via a "direct put" (API calls). Currently only Kinesis Data Streams and direct put are supported in the CDK.
See: Sending Data to a Delivery Stream in the Amazon Data Firehose Developer Guide.
Kinesis Data Stream
A delivery stream can read directly from a Kinesis data stream as a consumer of the data stream. Configure this behaviour by passing in a data stream in the source property via the KinesisStreamSource class when constructing a delivery stream:
ts
declare const destination: firehose.IDestination;
const sourceStream = new kinesis.Stream(this, 'Source Stream');
new firehose.DeliveryStream(this, 'Delivery Stream', {
source: new firehose.KinesisStreamSource(sourceStream),
destination: destination,
});
python
# destination: firehose.IDestination
source_stream = kinesis.Stream(self, "Source Stream")
firehose.DeliveryStream(self, "Delivery Stream",
source=firehose.KinesisStreamSource(source_stream),
destination=destination
)
java
IDestination destination;
Stream sourceStream = new Stream(this, "Source Stream");
DeliveryStream.Builder.create(this, "Delivery Stream")
.source(new KinesisStreamSource(sourceStream))
.destination(destination)
.build();
csharp
IDestination destination;
var sourceStream = new Stream(this, "Source Stream");
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Source = new KinesisStreamSource(sourceStream),
Destination = destination
});
go
var destination IDestination
sourceStream := kinesis.NewStream(this, jsii.String("Source Stream"))
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Source: firehose.NewKinesisStreamSource(sourceStream),
Destination: destination,
})
Direct Put
Data must be provided via "direct put", ie., by using a PutRecord or PutRecordBatch API call. There are a number of ways of doing so, such as:
- Kinesis Agent: a standalone Java application that monitors and delivers files while handling file rotation, checkpointing, and retries. See: Writing to Amazon Data Firehose Using Kinesis Agent in the Amazon Data Firehose Developer Guide.
- AWS SDK: a general purpose solution that allows you to deliver data to a delivery stream from anywhere using Java, .NET, Node.js, Python, or Ruby. See: Writing to Amazon Data Firehose Using the AWS SDK in the Amazon Data Firehose Developer Guide.
- CloudWatch Logs: subscribe to a log group and receive filtered log events directly into a delivery stream. See: logs-destinations.
- Eventbridge: add an event rule target to send events to a delivery stream based on the rule filtering. See: events-targets.
- SNS: add a subscription to send all notifications from the topic to a delivery stream. See: sns-subscriptions.
- IoT: add an action to an IoT rule to send various IoT information to a delivery stream
Destinations
Amazon Data Firehose supports multiple AWS and third-party services as destinations, including Amazon S3, Amazon Redshift, and more. You can find the full list of supported destination here.
Currently in the AWS CDK, S3, HTTP endpoint, and Datadog are implemented as L2 construct destinations. Other destinations can still be configured using L1 constructs.
HTTP
Defining a delivery stream with an HTTP destination:
ts
declare const endpointConfig: firehose.HttpEndpointConfig;
const httpDestination = new firehose.HttpEndpoint({
endpointConfig,
});
python
# endpoint_config: firehose.HttpEndpointConfig
http_destination = firehose.HttpEndpoint(
endpoint_config=endpoint_config
)
java
HttpEndpointConfig endpointConfig;
HttpEndpoint httpDestination = HttpEndpoint.Builder.create()
.endpointConfig(endpointConfig)
.build();
csharp
HttpEndpointConfig endpointConfig;
var httpDestination = new HttpEndpoint(new HttpEndpointProps {
EndpointConfig = endpointConfig
});
go
var endpointConfig HttpEndpointConfig
httpDestination := firehose.NewHttpEndpoint(&HttpEndpointProps{
EndpointConfig: EndpointConfig,
})
Datadog
Defining a delivery stream with a Datadog destination:
ts
import * as secretsmanager from 'aws-cdk-lib/aws-secretsmanager';
declare const apiKey: secretsmanager.Secret;
const datadogDestination = new firehose.Datadog({
apiKey,
endpoint: firehose.DatadogEndpoint.LOGS_US1,
});
python
import aws_cdk.aws_secretsmanager as secretsmanager
# api_key: secretsmanager.Secret
datadog_destination = firehose.Datadog(
api_key=api_key,
endpoint=firehose.DatadogEndpoint.LOGS_US1
)
java
import software.amazon.awscdk.services.secretsmanager.*;
Secret apiKey;
Datadog datadogDestination = Datadog.Builder.create()
.apiKey(apiKey)
.endpoint(DatadogEndpoint.LOGS_US1)
.build();
csharp
using Amazon.CDK.AWS.SecretsManager;
Secret apiKey;
var datadogDestination = new Datadog(new DatadogProps {
ApiKey = apiKey,
Endpoint = DatadogEndpoint.LOGS_US1
});
go
import secretsmanager "github.com/aws/aws-cdk-go/awscdk"
var apiKey Secret
datadogDestination := firehose.NewDatadog(&DatadogProps{
ApiKey: ApiKey,
Endpoint: firehose.DatadogEndpoint_LOGS_US1(),
})
S3
Defining a delivery stream with an S3 bucket destination:
ts
declare const bucket: s3.Bucket;
const s3Destination = new firehose.S3Bucket(bucket);
new firehose.DeliveryStream(this, 'Delivery Stream', {
destination: s3Destination,
});
python
# bucket: s3.Bucket
s3_destination = firehose.S3Bucket(bucket)
firehose.DeliveryStream(self, "Delivery Stream",
destination=s3_destination
)
java
Bucket bucket;
S3Bucket s3Destination = new S3Bucket(bucket);
DeliveryStream.Builder.create(this, "Delivery Stream")
.destination(s3Destination)
.build();
csharp
Bucket bucket;
var s3Destination = new S3Bucket(bucket);
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Destination = s3Destination
});
go
var bucket Bucket
s3Destination := firehose.NewS3Bucket(bucket)
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Destination: s3Destination,
})
The S3 destination also supports custom dynamic prefixes. dataOutputPrefix will be used for files successfully delivered to S3. errorOutputPrefix will be added to failed records before writing them to S3.
ts
import { TimeZone } from 'aws-cdk-lib';
declare const bucket: s3.Bucket;
const s3Destination = new firehose.S3Bucket(bucket, {
dataOutputPrefix: 'myFirehose/DeliveredYear=!{timestamp:yyyy}/anyMonth/rand=!{firehose:random-string}',
errorOutputPrefix: 'myFirehoseFailures/!{firehose:error-output-type}/!{timestamp:yyyy}/anyMonth/!{timestamp:dd}',
// The time zone of timestamps (default UTC)
timeZone: TimeZone.ASIA_TOKYO,
});
python
from aws_cdk import TimeZone
# bucket: s3.Bucket
s3_destination = firehose.S3Bucket(bucket,
data_output_prefix="myFirehose/DeliveredYear=!{timestamp:yyyy}/anyMonth/rand=!{firehose:random-string}",
error_output_prefix="myFirehoseFailures/!{firehose:error-output-type}/!{timestamp:yyyy}/anyMonth/!{timestamp:dd}",
# The time zone of timestamps (default UTC)
time_zone=TimeZone.ASIA_TOKYO
)
java
import software.amazon.awscdk.TimeZone;
Bucket bucket;
S3Bucket s3Destination = S3Bucket.Builder.create(bucket)
.dataOutputPrefix("myFirehose/DeliveredYear=!{timestamp:yyyy}/anyMonth/rand=!{firehose:random-string}")
.errorOutputPrefix("myFirehoseFailures/!{firehose:error-output-type}/!{timestamp:yyyy}/anyMonth/!{timestamp:dd}")
// The time zone of timestamps (default UTC)
.timeZone(TimeZone.ASIA_TOKYO)
.build();
csharp
using Amazon.CDK;
Bucket bucket;
var s3Destination = new S3Bucket(bucket, new S3BucketProps {
DataOutputPrefix = "myFirehose/DeliveredYear=!{timestamp:yyyy}/anyMonth/rand=!{firehose:random-string}",
ErrorOutputPrefix = "myFirehoseFailures/!{firehose:error-output-type}/!{timestamp:yyyy}/anyMonth/!{timestamp:dd}",
// The time zone of timestamps (default UTC)
TimeZone = TimeZone.ASIA_TOKYO
});
go
import "github.com/aws/aws-cdk-go/awscdk"
var bucket Bucket
s3Destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
DataOutputPrefix: jsii.String("myFirehose/DeliveredYear=!{timestamp:yyyy}/anyMonth/rand=!{firehose:random-string}"),
ErrorOutputPrefix: jsii.String("myFirehoseFailures/!{firehose:error-output-type}/!{timestamp:yyyy}/anyMonth/!{timestamp:dd}"),
// The time zone of timestamps (default UTC)
TimeZone: awscdk.TimeZone_ASIA_TOKYO(),
})
See: Custom S3 Prefixes in the Amazon Data Firehose Developer Guide.
To override default file extension appended by Data Format Conversion or S3 compression features, specify fileExtension.
ts
declare const bucket: s3.Bucket;
const s3Destination = new firehose.S3Bucket(bucket, {
compression: firehose.Compression.GZIP,
fileExtension: '.json.gz',
});
python
# bucket: s3.Bucket
s3_destination = firehose.S3Bucket(bucket,
compression=firehose.Compression.GZIP,
file_extension=".json.gz"
)
java
Bucket bucket;
S3Bucket s3Destination = S3Bucket.Builder.create(bucket)
.compression(Compression.GZIP)
.fileExtension(".json.gz")
.build();
csharp
Bucket bucket;
var s3Destination = new S3Bucket(bucket, new S3BucketProps {
Compression = Compression.GZIP,
FileExtension = ".json.gz"
});
go
var bucket Bucket
s3Destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
Compression: firehose.Compression_GZIP(),
FileExtension: jsii.String(".json.gz"),
})
Data Format Conversion
Data format conversion allows automatic conversion of inputs from JSON to either Parquet or ORC. Converting JSON records to columnar formats like Parquet or ORC can help speed up analytical querying while also increasing compression efficiency. When data format conversion is specified, it automatically enables Snappy compression on the output.
Only S3 Destinations support data format conversion.
An example of defining an S3 destination configured with data format conversion:
ts
declare const bucket: s3.Bucket;
declare const schemaGlueTable: glue.CfnTable;
const s3Destination = new firehose.S3Bucket(bucket, {
dataFormatConversion: {
schemaConfiguration: firehose.SchemaConfiguration.fromCfnTable(schemaGlueTable),
inputFormat: firehose.InputFormat.OPENX_JSON,
outputFormat: firehose.OutputFormat.PARQUET,
}
});
python
# bucket: s3.Bucket
# schema_glue_table: glue.CfnTable
s3_destination = firehose.S3Bucket(bucket,
data_format_conversion=firehose.DataFormatConversionProps(
schema_configuration=firehose.SchemaConfiguration.from_cfn_table(schema_glue_table),
input_format=firehose.InputFormat.OPENX_JSON,
output_format=firehose.OutputFormat.PARQUET
)
)
java
Bucket bucket;
CfnTable schemaGlueTable;
S3Bucket s3Destination = S3Bucket.Builder.create(bucket)
.dataFormatConversion(DataFormatConversionProps.builder()
.schemaConfiguration(SchemaConfiguration.fromCfnTable(schemaGlueTable))
.inputFormat(InputFormat.OPENX_JSON)
.outputFormat(OutputFormat.PARQUET)
.build())
.build();
csharp
Bucket bucket;
CfnTable schemaGlueTable;
var s3Destination = new S3Bucket(bucket, new S3BucketProps {
DataFormatConversion = new DataFormatConversionProps {
SchemaConfiguration = SchemaConfiguration.FromCfnTable(schemaGlueTable),
InputFormat = InputFormat.OPENX_JSON,
OutputFormat = OutputFormat.PARQUET
}
});
go
var bucket Bucket
var schemaGlueTable CfnTable
s3Destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
DataFormatConversion: &DataFormatConversionProps{
SchemaConfiguration: firehose.SchemaConfiguration_FromCfnTable(schemaGlueTable),
InputFormat: firehose.InputFormat_OPENX_JSON(),
OutputFormat: firehose.OutputFormat_PARQUET(),
},
})
When data format conversion is enabled, the Delivery Stream's buffering size must be at least 64 MiB. Additionally, the default buffering size is changed from 5 MiB to 128 MiB. This mirrors the Cloudformation behavior.
You can only parse JSON and transform it into either Parquet or ORC:
- to read JSON using OpenX parser, choose
InputFormat.OPENX_JSON. - to read JSON using Hive parser, choose
InputFormat.HIVE_JSON. - to transform into Parquet, choose
OutputFormat.PARQUET. - to transform into ORC, choose
OutputFormat.ORC.
The following subsections explain how to specify advanced configuration options for each input and output format if the defaults are not desirable
Input Format: OpenX JSON
Example creation of custom OpenX JSON InputFormat:
ts
const inputFormat = new firehose.OpenXJsonInputFormat({
lowercaseColumnNames: false,
columnToJsonKeyMappings: {"ts": "timestamp"},
convertDotsInJsonKeysToUnderscores: true,
})
python
input_format = firehose.OpenXJsonInputFormat(
lowercase_column_names=False,
column_to_json_key_mappings={"ts": "timestamp"},
convert_dots_in_json_keys_to_underscores=True
)
java
OpenXJsonInputFormat inputFormat = OpenXJsonInputFormat.Builder.create()
.lowercaseColumnNames(false)
.columnToJsonKeyMappings(Map.of("ts", "timestamp"))
.convertDotsInJsonKeysToUnderscores(true)
.build();
csharp
var inputFormat = new OpenXJsonInputFormat(new OpenXJsonInputFormatProps {
LowercaseColumnNames = false,
ColumnToJsonKeyMappings = new Dictionary<string, string> { { "ts", "timestamp" } },
ConvertDotsInJsonKeysToUnderscores = true
});
go
inputFormat := firehose.NewOpenXJsonInputFormat(&OpenXJsonInputFormatProps{
LowercaseColumnNames: jsii.Boolean(false),
ColumnToJsonKeyMappings: map[string]*string{
"ts": jsii.String("timestamp"),
},
ConvertDotsInJsonKeysToUnderscores: jsii.Boolean(true),
})
Input Format: Hive JSON
Example creation of custom Hive JSON InputFormat:
ts
const inputFormat = new firehose.HiveJsonInputFormat({
timestampParsers: [
firehose.TimestampParser.fromFormatString('yyyy-MM-dd'),
firehose.TimestampParser.EPOCH_MILLIS,
]
})
python
input_format = firehose.HiveJsonInputFormat(
timestamp_parsers=[
firehose.TimestampParser.from_format_string("yyyy-MM-dd"), firehose.TimestampParser.EPOCH_MILLIS
]
)
java
HiveJsonInputFormat inputFormat = HiveJsonInputFormat.Builder.create()
.timestampParsers(List.of(TimestampParser.fromFormatString("yyyy-MM-dd"), TimestampParser.EPOCH_MILLIS))
.build();
csharp
var inputFormat = new HiveJsonInputFormat(new HiveJsonInputFormatProps {
TimestampParsers = new [] { TimestampParser.FromFormatString("yyyy-MM-dd"), TimestampParser.EPOCH_MILLIS }
});
go
inputFormat := firehose.NewHiveJsonInputFormat(&HiveJsonInputFormatProps{
TimestampParsers: []TimestampParser{
firehose.TimestampParser_FromFormatString(jsii.String("yyyy-MM-dd")),
firehose.TimestampParser_EPOCH_MILLIS(),
},
})
Hive JSON allows you to specify custom timestamp formats to parse. The syntax of the format string is Joda Time.
To parse timestamps formatted as milliseconds since epoch, use the convenience constant TimestampParser.EPOCH_MILLIS.
Output Format: Parquet
Example of a custom Parquet OutputFormat, with all values changed from the defaults.
ts
const outputFormat = new firehose.ParquetOutputFormat({
blockSize: Size.mebibytes(512),
compression: firehose.ParquetCompression.UNCOMPRESSED,
enableDictionaryCompression: true,
maxPadding: Size.bytes(10),
pageSize: Size.mebibytes(2),
writerVersion: firehose.ParquetWriterVersion.V2,
})
python
output_format = firehose.ParquetOutputFormat(
block_size=Size.mebibytes(512),
compression=firehose.ParquetCompression.UNCOMPRESSED,
enable_dictionary_compression=True,
max_padding=Size.bytes(10),
page_size=Size.mebibytes(2),
writer_version=firehose.ParquetWriterVersion.V2
)
java
ParquetOutputFormat outputFormat = ParquetOutputFormat.Builder.create()
.blockSize(Size.mebibytes(512))
.compression(ParquetCompression.UNCOMPRESSED)
.enableDictionaryCompression(true)
.maxPadding(Size.bytes(10))
.pageSize(Size.mebibytes(2))
.writerVersion(ParquetWriterVersion.V2)
.build();
csharp
var outputFormat = new ParquetOutputFormat(new ParquetOutputFormatProps {
BlockSize = Size.Mebibytes(512),
Compression = ParquetCompression.UNCOMPRESSED,
EnableDictionaryCompression = true,
MaxPadding = Size.Bytes(10),
PageSize = Size.Mebibytes(2),
WriterVersion = ParquetWriterVersion.V2
});
go
outputFormat := firehose.NewParquetOutputFormat(&ParquetOutputFormatProps{
BlockSize: awscdk.Size_Mebibytes(jsii.Number(512)),
Compression: firehose.ParquetCompression_UNCOMPRESSED(),
EnableDictionaryCompression: jsii.Boolean(true),
MaxPadding: awscdk.Size_Bytes(jsii.Number(10)),
PageSize: awscdk.Size_*Mebibytes(jsii.Number(2)),
WriterVersion: firehose.ParquetWriterVersion_V2,
})
Output Format: ORC
Example creation of custom ORC OutputFormat, with all values changed from the defaults.
ts
const outputFormat = new firehose.OrcOutputFormat({
formatVersion: firehose.OrcFormatVersion.V0_11,
blockSize: Size.mebibytes(256),
compression: firehose.OrcCompression.NONE,
bloomFilterColumns: ['columnA'],
bloomFilterFalsePositiveProbability: 0.1,
dictionaryKeyThreshold: 0.7,
enablePadding: true,
paddingTolerance: 0.2,
rowIndexStride: 9000,
stripeSize: Size.mebibytes(32),
})
python
output_format = firehose.OrcOutputFormat(
format_version=firehose.OrcFormatVersion.V0_11,
block_size=Size.mebibytes(256),
compression=firehose.OrcCompression.NONE,
bloom_filter_columns=["columnA"],
bloom_filter_false_positive_probability=0.1,
dictionary_key_threshold=0.7,
enable_padding=True,
padding_tolerance=0.2,
row_index_stride=9000,
stripe_size=Size.mebibytes(32)
)
java
OrcOutputFormat outputFormat = OrcOutputFormat.Builder.create()
.formatVersion(OrcFormatVersion.V0_11)
.blockSize(Size.mebibytes(256))
.compression(OrcCompression.NONE)
.bloomFilterColumns(List.of("columnA"))
.bloomFilterFalsePositiveProbability(0.1)
.dictionaryKeyThreshold(0.7)
.enablePadding(true)
.paddingTolerance(0.2)
.rowIndexStride(9000)
.stripeSize(Size.mebibytes(32))
.build();
csharp
var outputFormat = new OrcOutputFormat(new OrcOutputFormatProps {
FormatVersion = OrcFormatVersion.V0_11,
BlockSize = Size.Mebibytes(256),
Compression = OrcCompression.NONE,
BloomFilterColumns = new [] { "columnA" },
BloomFilterFalsePositiveProbability = 0.1,
DictionaryKeyThreshold = 0.7,
EnablePadding = true,
PaddingTolerance = 0.2,
RowIndexStride = 9000,
StripeSize = Size.Mebibytes(32)
});
go
outputFormat := firehose.NewOrcOutputFormat(&OrcOutputFormatProps{
FormatVersion: firehose.OrcFormatVersion_V0_11,
BlockSize: awscdk.Size_Mebibytes(jsii.Number(256)),
Compression: firehose.OrcCompression_NONE(),
BloomFilterColumns: []*string{
jsii.String("columnA"),
},
BloomFilterFalsePositiveProbability: jsii.Number(0.1),
DictionaryKeyThreshold: jsii.Number(0.7),
EnablePadding: jsii.Boolean(true),
PaddingTolerance: jsii.Number(0.2),
RowIndexStride: jsii.Number(9000),
StripeSize: awscdk.Size_*Mebibytes(jsii.Number(32)),
})
Server-side Encryption
Enabling server-side encryption (SSE) requires Amazon Data Firehose to encrypt all data sent to delivery stream when it is stored at rest. This means that data is encrypted before being written to the service's internal storage layer and decrypted after it is received from the internal storage layer. The service manages keys and cryptographic operations so that sources and destinations do not need to, as the data is encrypted and decrypted at the boundaries of the service (i.e., before the data is delivered to a destination). By default, delivery streams do not have SSE enabled.
The Key Management Service keys (KMS keys) used for SSE can either be AWS-owned or customer-managed. AWS-owned KMS keys are created, owned and managed by AWS for use in multiple AWS accounts. As a customer, you cannot view, use, track, or manage these keys, and you are not charged for their use. On the other hand, customer-managed KMS keys are created and owned within your account and managed entirely by you. As a customer, you are responsible for managing access, rotation, aliases, and deletion for these keys, and you are changed for their use.
See: AWS KMS keys in the KMS Developer Guide.
ts
declare const destination: firehose.IDestination;
// SSE with an AWS-owned key
new firehose.DeliveryStream(this, 'Delivery Stream with AWS Owned Key', {
encryption: firehose.StreamEncryption.awsOwnedKey(),
destination: destination,
});
// SSE with an customer-managed key that is created automatically by the CDK
new firehose.DeliveryStream(this, 'Delivery Stream with Customer Managed Key', {
encryption: firehose.StreamEncryption.customerManagedKey(),
destination: destination,
});
// SSE with an customer-managed key that is explicitly specified
declare const key: kms.Key;
new firehose.DeliveryStream(this, 'Delivery Stream with Customer Managed and Provided Key', {
encryption: firehose.StreamEncryption.customerManagedKey(key),
destination: destination,
});
python
# destination: firehose.IDestination
# SSE with an customer-managed key that is explicitly specified
# key: kms.Key
# SSE with an AWS-owned key
firehose.DeliveryStream(self, "Delivery Stream with AWS Owned Key",
encryption=firehose.StreamEncryption.aws_owned_key(),
destination=destination
)
# SSE with an customer-managed key that is created automatically by the CDK
firehose.DeliveryStream(self, "Delivery Stream with Customer Managed Key",
encryption=firehose.StreamEncryption.customer_managed_key(),
destination=destination
)
firehose.DeliveryStream(self, "Delivery Stream with Customer Managed and Provided Key",
encryption=firehose.StreamEncryption.customer_managed_key(key),
destination=destination
)
java
IDestination destination;
// SSE with an customer-managed key that is explicitly specified
Key key;
// SSE with an AWS-owned key
// SSE with an AWS-owned key
DeliveryStream.Builder.create(this, "Delivery Stream with AWS Owned Key")
.encryption(StreamEncryption.awsOwnedKey())
.destination(destination)
.build();
// SSE with an customer-managed key that is created automatically by the CDK
// SSE with an customer-managed key that is created automatically by the CDK
DeliveryStream.Builder.create(this, "Delivery Stream with Customer Managed Key")
.encryption(StreamEncryption.customerManagedKey())
.destination(destination)
.build();
DeliveryStream.Builder.create(this, "Delivery Stream with Customer Managed and Provided Key")
.encryption(StreamEncryption.customerManagedKey(key))
.destination(destination)
.build();
csharp
IDestination destination;
// SSE with an customer-managed key that is explicitly specified
Key key;
// SSE with an AWS-owned key
// SSE with an AWS-owned key
new DeliveryStream(this, "Delivery Stream with AWS Owned Key", new DeliveryStreamProps {
Encryption = StreamEncryption.AwsOwnedKey(),
Destination = destination
});
// SSE with an customer-managed key that is created automatically by the CDK
// SSE with an customer-managed key that is created automatically by the CDK
new DeliveryStream(this, "Delivery Stream with Customer Managed Key", new DeliveryStreamProps {
Encryption = StreamEncryption.CustomerManagedKey(),
Destination = destination
});
new DeliveryStream(this, "Delivery Stream with Customer Managed and Provided Key", new DeliveryStreamProps {
Encryption = StreamEncryption.CustomerManagedKey(key),
Destination = destination
});
go
var destination IDestination
// SSE with an customer-managed key that is explicitly specified
var key Key
// SSE with an AWS-owned key
// SSE with an AWS-owned key
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream with AWS Owned Key"), &DeliveryStreamProps{
Encryption: firehose.StreamEncryption_AwsOwnedKey(),
Destination: destination,
})
// SSE with an customer-managed key that is created automatically by the CDK
// SSE with an customer-managed key that is created automatically by the CDK
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream with Customer Managed Key"), &DeliveryStreamProps{
Encryption: firehose.StreamEncryption_CustomerManagedKey(),
Destination: destination,
})
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream with Customer Managed and Provided Key"), &DeliveryStreamProps{
Encryption: firehose.StreamEncryption_*CustomerManagedKey(key),
Destination: destination,
})
See: Data Protection in the Amazon Data Firehose Developer Guide.
Monitoring
Amazon Data Firehose is integrated with CloudWatch, so you can monitor the performance of your delivery streams via logs and metrics.
Logs
Amazon Data Firehose will send logs to CloudWatch when data transformation or data delivery fails. The CDK will enable logging by default and create a CloudWatch LogGroup and LogStream with default settings for your Delivery Stream.
When creating a destination, you can provide an ILoggingConfig, which can either be an EnableLogging or DisableLogging instance. If you use EnableLogging, the CDK will create a CloudWatch LogGroup and LogStream with all CloudFormation default settings for you, or you can optionally specify your own log group to be used for capturing and storing log events. For example:
ts
import * as logs from 'aws-cdk-lib/aws-logs';
const logGroup = new logs.LogGroup(this, 'Log Group');
declare const bucket: s3.Bucket;
const destination = new firehose.S3Bucket(bucket, {
loggingConfig: new firehose.EnableLogging(logGroup),
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
destination: destination,
});
python
import aws_cdk.aws_logs as logs
# bucket: s3.Bucket
log_group = logs.LogGroup(self, "Log Group")
destination = firehose.S3Bucket(bucket,
logging_config=firehose.EnableLogging(log_group)
)
firehose.DeliveryStream(self, "Delivery Stream",
destination=destination
)
java
import software.amazon.awscdk.services.logs.*;
Bucket bucket;
LogGroup logGroup = new LogGroup(this, "Log Group");
S3Bucket destination = S3Bucket.Builder.create(bucket)
.loggingConfig(new EnableLogging(logGroup))
.build();
DeliveryStream.Builder.create(this, "Delivery Stream")
.destination(destination)
.build();
csharp
using Amazon.CDK.AWS.Logs;
Bucket bucket;
var logGroup = new LogGroup(this, "Log Group");
var destination = new S3Bucket(bucket, new S3BucketProps {
LoggingConfig = new EnableLogging(logGroup)
});
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Destination = destination
});
go
import logs "github.com/aws/aws-cdk-go/awscdk"
var bucket Bucket
logGroup := logs.NewLogGroup(this, jsii.String("Log Group"))
destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
LoggingConfig: firehose.NewEnableLogging(logGroup),
})
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Destination: destination,
})
Logging can also be disabled:
ts
declare const bucket: s3.Bucket;
const destination = new firehose.S3Bucket(bucket, {
loggingConfig: new firehose.DisableLogging(),
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
destination: destination,
});
python
# bucket: s3.Bucket
destination = firehose.S3Bucket(bucket,
logging_config=firehose.DisableLogging()
)
firehose.DeliveryStream(self, "Delivery Stream",
destination=destination
)
java
Bucket bucket;
S3Bucket destination = S3Bucket.Builder.create(bucket)
.loggingConfig(new DisableLogging())
.build();
DeliveryStream.Builder.create(this, "Delivery Stream")
.destination(destination)
.build();
csharp
Bucket bucket;
var destination = new S3Bucket(bucket, new S3BucketProps {
LoggingConfig = new DisableLogging()
});
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Destination = destination
});
go
var bucket Bucket
destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
LoggingConfig: firehose.NewDisableLogging(),
})
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Destination: destination,
})
See: Monitoring using CloudWatch Logs in the Amazon Data Firehose Developer Guide.
Metrics
Amazon Data Firehose sends metrics to CloudWatch so that you can collect and analyze the performance of the delivery stream, including data delivery, data ingestion, data transformation, format conversion, API usage, encryption, and resource usage. You can then use CloudWatch alarms to alert you, for example, when data freshness (the age of the oldest record in the delivery stream) exceeds the buffering limit (indicating that data is not being delivered to your destination), or when the rate of incoming records exceeds the limit of records per second (indicating data is flowing into your delivery stream faster than it is configured to process).
CDK provides methods for accessing delivery stream metrics with default configuration, such as metricIncomingBytes, and metricIncomingRecords (see IDeliveryStream for a full list). CDK also provides a generic metric method that can be used to produce metric configurations for any metric provided by Amazon Data Firehose; the configurations are pre-populated with the correct dimensions for the delivery stream.
ts
import * as cloudwatch from 'aws-cdk-lib/aws-cloudwatch';
declare const deliveryStream: firehose.DeliveryStream;
// Alarm that triggers when the per-second average of incoming bytes exceeds 90% of the current service limit
const incomingBytesPercentOfLimit = new cloudwatch.MathExpression({
expression: 'incomingBytes / 300 / bytePerSecLimit',
usingMetrics: {
incomingBytes: deliveryStream.metricIncomingBytes({ statistic: cloudwatch.Statistic.SUM }),
bytePerSecLimit: deliveryStream.metric('BytesPerSecondLimit'),
},
});
new cloudwatch.Alarm(this, 'Alarm', {
metric: incomingBytesPercentOfLimit,
threshold: 0.9,
evaluationPeriods: 3,
});
python
import aws_cdk.aws_cloudwatch as cloudwatch
# delivery_stream: firehose.DeliveryStream
# Alarm that triggers when the per-second average of incoming bytes exceeds 90% of the current service limit
incoming_bytes_percent_of_limit = cloudwatch.MathExpression(
expression="incomingBytes / 300 / bytePerSecLimit",
using_metrics={
"incoming_bytes": delivery_stream.metric_incoming_bytes(statistic=cloudwatch.Statistic.SUM),
"byte_per_sec_limit": delivery_stream.metric("BytesPerSecondLimit")
}
)
cloudwatch.Alarm(self, "Alarm",
metric=incoming_bytes_percent_of_limit,
threshold=0.9,
evaluation_periods=3
)
java
import software.amazon.awscdk.services.cloudwatch.*;
DeliveryStream deliveryStream;
// Alarm that triggers when the per-second average of incoming bytes exceeds 90% of the current service limit
MathExpression incomingBytesPercentOfLimit = MathExpression.Builder.create()
.expression("incomingBytes / 300 / bytePerSecLimit")
.usingMetrics(Map.of(
"incomingBytes", deliveryStream.metricIncomingBytes(MetricOptions.builder().statistic(Statistic.SUM).build()),
"bytePerSecLimit", deliveryStream.metric("BytesPerSecondLimit")))
.build();
Alarm.Builder.create(this, "Alarm")
.metric(incomingBytesPercentOfLimit)
.threshold(0.9)
.evaluationPeriods(3)
.build();
csharp
using Amazon.CDK.AWS.CloudWatch;
DeliveryStream deliveryStream;
// Alarm that triggers when the per-second average of incoming bytes exceeds 90% of the current service limit
var incomingBytesPercentOfLimit = new MathExpression(new MathExpressionProps {
Expression = "incomingBytes / 300 / bytePerSecLimit",
UsingMetrics = new Dictionary<string, IMetric> {
{ "incomingBytes", deliveryStream.MetricIncomingBytes(new MetricOptions { Statistic = Statistic.SUM }) },
{ "bytePerSecLimit", deliveryStream.Metric("BytesPerSecondLimit") }
}
});
new Alarm(this, "Alarm", new AlarmProps {
Metric = incomingBytesPercentOfLimit,
Threshold = 0.9,
EvaluationPeriods = 3
});
go
import "github.com/aws/aws-cdk-go/awscdk"
var deliveryStream DeliveryStream
// Alarm that triggers when the per-second average of incoming bytes exceeds 90% of the current service limit
incomingBytesPercentOfLimit := cloudwatch.NewMathExpression(&MathExpressionProps{
Expression: jsii.String("incomingBytes / 300 / bytePerSecLimit"),
UsingMetrics: map[string]IMetric{
"incomingBytes": deliveryStream.metricIncomingBytes(&MetricOptions{
"statistic": cloudwatch.Statistic_SUM,
}),
"bytePerSecLimit": deliveryStream.metric(jsii.String("BytesPerSecondLimit")),
},
})
cloudwatch.NewAlarm(this, jsii.String("Alarm"), &AlarmProps{
Metric: incomingBytesPercentOfLimit,
Threshold: jsii.Number(0.9),
EvaluationPeriods: jsii.Number(3),
})
See: Monitoring Using CloudWatch Metrics in the Amazon Data Firehose Developer Guide.
Compression
Your data can automatically be compressed when it is delivered to S3 as either a final or an intermediary/backup destination. Supported compression formats are: gzip, Snappy, Hadoop-compatible Snappy, and ZIP, except for Redshift destinations, where Snappy (regardless of Hadoop-compatibility) and ZIP are not supported. By default, data is delivered to S3 without compression.
ts
// Compress data delivered to S3 using Snappy
declare const bucket: s3.Bucket;
const s3Destination = new firehose.S3Bucket(bucket, {
compression: firehose.Compression.SNAPPY,
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
destination: s3Destination,
});
python
# Compress data delivered to S3 using Snappy
# bucket: s3.Bucket
s3_destination = firehose.S3Bucket(bucket,
compression=firehose.Compression.SNAPPY
)
firehose.DeliveryStream(self, "Delivery Stream",
destination=s3_destination
)
java
// Compress data delivered to S3 using Snappy
Bucket bucket;
S3Bucket s3Destination = S3Bucket.Builder.create(bucket)
.compression(Compression.SNAPPY)
.build();
DeliveryStream.Builder.create(this, "Delivery Stream")
.destination(s3Destination)
.build();
csharp
// Compress data delivered to S3 using Snappy
Bucket bucket;
var s3Destination = new S3Bucket(bucket, new S3BucketProps {
Compression = Compression.SNAPPY
});
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Destination = s3Destination
});
go
// Compress data delivered to S3 using Snappy
var bucket Bucket
s3Destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
Compression: firehose.Compression_SNAPPY(),
})
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Destination: s3Destination,
})
Buffering
Incoming data is buffered before it is delivered to the specified destination. The delivery stream will wait until the amount of incoming data has exceeded some threshold (the "buffer size") or until the time since the last data delivery occurred exceeds some threshold (the "buffer interval"), whichever happens first. You can configure these thresholds based on the capabilities of the destination and your use-case. By default, the buffer size is 5 MiB and the buffer interval is 5 minutes.
ts
// Increase the buffer interval and size to 10 minutes and 8 MiB, respectively
declare const bucket: s3.Bucket;
const destination = new firehose.S3Bucket(bucket, {
bufferingInterval: Duration.minutes(10),
bufferingSize: Size.mebibytes(8),
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
destination: destination,
});
python
# Increase the buffer interval and size to 10 minutes and 8 MiB, respectively
# bucket: s3.Bucket
destination = firehose.S3Bucket(bucket,
buffering_interval=Duration.minutes(10),
buffering_size=Size.mebibytes(8)
)
firehose.DeliveryStream(self, "Delivery Stream",
destination=destination
)
java
// Increase the buffer interval and size to 10 minutes and 8 MiB, respectively
Bucket bucket;
S3Bucket destination = S3Bucket.Builder.create(bucket)
.bufferingInterval(Duration.minutes(10))
.bufferingSize(Size.mebibytes(8))
.build();
DeliveryStream.Builder.create(this, "Delivery Stream")
.destination(destination)
.build();
csharp
// Increase the buffer interval and size to 10 minutes and 8 MiB, respectively
Bucket bucket;
var destination = new S3Bucket(bucket, new S3BucketProps {
BufferingInterval = Duration.Minutes(10),
BufferingSize = Size.Mebibytes(8)
});
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Destination = destination
});
go
// Increase the buffer interval and size to 10 minutes and 8 MiB, respectively
var bucket Bucket
destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
BufferingInterval: awscdk.Duration_Minutes(jsii.Number(10)),
BufferingSize: awscdk.Size_Mebibytes(jsii.Number(8)),
})
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Destination: destination,
})
See: Data Delivery Frequency in the Amazon Data Firehose Developer Guide.
Zero buffering, where Amazon Data Firehose stream can be configured to not buffer data before delivery, is supported by setting the "buffer interval" to 0.
ts
// Setup zero buffering
declare const bucket: s3.Bucket;
const destination = new firehose.S3Bucket(bucket, {
bufferingInterval: Duration.seconds(0),
});
new firehose.DeliveryStream(this, 'ZeroBufferDeliveryStream', {
destination: destination,
});
python
# Setup zero buffering
# bucket: s3.Bucket
destination = firehose.S3Bucket(bucket,
buffering_interval=Duration.seconds(0)
)
firehose.DeliveryStream(self, "ZeroBufferDeliveryStream",
destination=destination
)
java
// Setup zero buffering
Bucket bucket;
S3Bucket destination = S3Bucket.Builder.create(bucket)
.bufferingInterval(Duration.seconds(0))
.build();
DeliveryStream.Builder.create(this, "ZeroBufferDeliveryStream")
.destination(destination)
.build();
csharp
// Setup zero buffering
Bucket bucket;
var destination = new S3Bucket(bucket, new S3BucketProps {
BufferingInterval = Duration.Seconds(0)
});
new DeliveryStream(this, "ZeroBufferDeliveryStream", new DeliveryStreamProps {
Destination = destination
});
go
// Setup zero buffering
var bucket Bucket
destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
BufferingInterval: awscdk.Duration_Seconds(jsii.Number(0)),
})
firehose.NewDeliveryStream(this, jsii.String("ZeroBufferDeliveryStream"), &DeliveryStreamProps{
Destination: destination,
})
See: Buffering Hints.
Destination Encryption
Your data can be automatically encrypted when it is delivered to S3 as a final or an intermediary/backup destination. Amazon Data Firehose supports Amazon S3 server-side encryption with AWS Key Management Service (AWS KMS) for encrypting delivered data in Amazon S3. You can choose to not encrypt the data or to encrypt with a key from the list of AWS KMS keys that you own. For more information, see Protecting Data Using Server-Side Encryption with AWS KMS–Managed Keys (SSE-KMS). By default, encryption isn’t directly enabled on the delivery stream; instead, it uses the default encryption settings of the destination S3 bucket.
ts
declare const bucket: s3.Bucket;
declare const key: kms.Key;
const destination = new firehose.S3Bucket(bucket, {
encryptionKey: key,
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
destination: destination,
});
python
# bucket: s3.Bucket
# key: kms.Key
destination = firehose.S3Bucket(bucket,
encryption_key=key
)
firehose.DeliveryStream(self, "Delivery Stream",
destination=destination
)
java
Bucket bucket;
Key key;
S3Bucket destination = S3Bucket.Builder.create(bucket)
.encryptionKey(key)
.build();
DeliveryStream.Builder.create(this, "Delivery Stream")
.destination(destination)
.build();
csharp
Bucket bucket;
Key key;
var destination = new S3Bucket(bucket, new S3BucketProps {
EncryptionKey = key
});
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Destination = destination
});
go
var bucket Bucket
var key Key
destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
EncryptionKey: key,
})
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Destination: destination,
})
Backup
A delivery stream can be configured to back up data to S3 that it attempted to deliver to the configured destination. Backed up data can be all the data that the delivery stream attempted to deliver or just data that it failed to deliver (Redshift and S3 destinations can only back up all data). CDK can create a new S3 bucket where it will back up data, or you can provide a bucket where data will be backed up. You can also provide a prefix under which your backed-up data will be placed within the bucket. By default, source data is not backed up to S3.
ts
// Enable backup of all source records (to an S3 bucket created by CDK).
declare const bucket: s3.Bucket;
new firehose.DeliveryStream(this, 'Delivery Stream Backup All', {
destination:
new firehose.S3Bucket(bucket, {
s3Backup: {
mode: firehose.BackupMode.ALL,
},
}),
});
// Explicitly provide an S3 bucket to which all source records will be backed up.
declare const backupBucket: s3.Bucket;
new firehose.DeliveryStream(this, 'Delivery Stream Backup All Explicit Bucket', {
destination:
new firehose.S3Bucket(bucket, {
s3Backup: {
bucket: backupBucket,
},
}),
});
// Explicitly provide an S3 prefix under which all source records will be backed up.
new firehose.DeliveryStream(this, 'Delivery Stream Backup All Explicit Prefix', {
destination:
new firehose.S3Bucket(bucket, {
s3Backup: {
mode: firehose.BackupMode.ALL,
dataOutputPrefix: 'mybackup',
},
}),
});
python
# Enable backup of all source records (to an S3 bucket created by CDK).
# bucket: s3.Bucket
# Explicitly provide an S3 bucket to which all source records will be backed up.
# backup_bucket: s3.Bucket
firehose.DeliveryStream(self, "Delivery Stream Backup All",
destination=
firehose.S3Bucket(bucket,
s3_backup=firehose.DestinationS3BackupProps(
mode=firehose.BackupMode.ALL
)
)
)
firehose.DeliveryStream(self, "Delivery Stream Backup All Explicit Bucket",
destination=
firehose.S3Bucket(bucket,
s3_backup=firehose.DestinationS3BackupProps(
bucket=backup_bucket
)
)
)
# Explicitly provide an S3 prefix under which all source records will be backed up.
firehose.DeliveryStream(self, "Delivery Stream Backup All Explicit Prefix",
destination=
firehose.S3Bucket(bucket,
s3_backup=firehose.DestinationS3BackupProps(
mode=firehose.BackupMode.ALL,
data_output_prefix="mybackup"
)
)
)
java
// Enable backup of all source records (to an S3 bucket created by CDK).
Bucket bucket;
// Explicitly provide an S3 bucket to which all source records will be backed up.
Bucket backupBucket;
DeliveryStream.Builder.create(this, "Delivery Stream Backup All")
.destination(
S3Bucket.Builder.create(bucket)
.s3Backup(DestinationS3BackupProps.builder()
.mode(BackupMode.ALL)
.build())
.build())
.build();
DeliveryStream.Builder.create(this, "Delivery Stream Backup All Explicit Bucket")
.destination(
S3Bucket.Builder.create(bucket)
.s3Backup(DestinationS3BackupProps.builder()
.bucket(backupBucket)
.build())
.build())
.build();
// Explicitly provide an S3 prefix under which all source records will be backed up.
// Explicitly provide an S3 prefix under which all source records will be backed up.
DeliveryStream.Builder.create(this, "Delivery Stream Backup All Explicit Prefix")
.destination(
S3Bucket.Builder.create(bucket)
.s3Backup(DestinationS3BackupProps.builder()
.mode(BackupMode.ALL)
.dataOutputPrefix("mybackup")
.build())
.build())
.build();
csharp
// Enable backup of all source records (to an S3 bucket created by CDK).
Bucket bucket;
// Explicitly provide an S3 bucket to which all source records will be backed up.
Bucket backupBucket;
new DeliveryStream(this, "Delivery Stream Backup All", new DeliveryStreamProps {
Destination =
new S3Bucket(bucket, new S3BucketProps {
S3Backup = new DestinationS3BackupProps {
Mode = BackupMode.ALL
}
})
});
new DeliveryStream(this, "Delivery Stream Backup All Explicit Bucket", new DeliveryStreamProps {
Destination =
new S3Bucket(bucket, new S3BucketProps {
S3Backup = new DestinationS3BackupProps {
Bucket = backupBucket
}
})
});
// Explicitly provide an S3 prefix under which all source records will be backed up.
// Explicitly provide an S3 prefix under which all source records will be backed up.
new DeliveryStream(this, "Delivery Stream Backup All Explicit Prefix", new DeliveryStreamProps {
Destination =
new S3Bucket(bucket, new S3BucketProps {
S3Backup = new DestinationS3BackupProps {
Mode = BackupMode.ALL,
DataOutputPrefix = "mybackup"
}
})
});
go
// Enable backup of all source records (to an S3 bucket created by CDK).
var bucket Bucket
// Explicitly provide an S3 bucket to which all source records will be backed up.
var backupBucket Bucket
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream Backup All"), &DeliveryStreamProps{
Destination:
firehose.NewS3Bucket(bucket, &S3BucketProps{
S3Backup: &DestinationS3BackupProps{
Mode: firehose.BackupMode_ALL,
},
}),
})
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream Backup All Explicit Bucket"), &DeliveryStreamProps{
Destination:
firehose.NewS3Bucket(bucket, &S3BucketProps{
S3Backup: &DestinationS3BackupProps{
Bucket: backupBucket,
},
}),
})
// Explicitly provide an S3 prefix under which all source records will be backed up.
// Explicitly provide an S3 prefix under which all source records will be backed up.
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream Backup All Explicit Prefix"), &DeliveryStreamProps{
Destination:
firehose.NewS3Bucket(bucket, &S3BucketProps{
S3Backup: &DestinationS3BackupProps{
Mode: firehose.BackupMode_ALL,
DataOutputPrefix: jsii.String("mybackup"),
},
}),
})
If any Data Processing or Transformation is configured on your Delivery Stream, the source records will be backed up in their original format.
Data Processing/Transformation
Data can be transformed before being delivered to destinations. There are two types of data processing for delivery streams: record transformation with AWS Lambda, and record format conversion using a schema stored in an AWS Glue table. If both types of data processing are configured, then the Lambda transformation is performed first. By default, no data processing occurs.
This construct library currently only supports data transformation with AWS Lambda and some built-in data processors. See #15501 to track the status of adding support for record format conversion.
Data transformation with AWS Lambda
To transform the data, Amazon Data Firehose will call a Lambda function that you provide and deliver the data returned in place of the source record. The function must return a result that contains records in a specific format, including the following fields:
recordId-- the ID of the input record that corresponds the results.result-- the status of the transformation of the record: "Ok" (success), "Dropped" (not processed intentionally), or "ProcessingFailed" (not processed due to an error).data-- the transformed data, Base64-encoded.metadata-- the metadata used by dynamic partitioning.
The data is buffered up to 1 minute and up to 3 MiB by default before being sent to the function, but can be configured using bufferInterval and bufferSize in the processor configuration (see: Buffering). If the function invocation fails due to a network timeout or because of hitting an invocation limit, the invocation is retried 3 times by default, but can be configured using retries in the processor configuration.
ts
// Provide a Lambda function that will transform records before delivery, with custom
// buffering and retry configuration
const lambdaFunction = new lambda.Function(this, 'Processor', {
runtime: lambda.Runtime.NODEJS_LATEST,
handler: 'index.handler',
code: lambda.Code.fromAsset(path.join(__dirname, 'process-records')),
});
const lambdaProcessor = new firehose.LambdaFunctionProcessor(lambdaFunction, {
bufferInterval: Duration.minutes(5),
bufferSize: Size.mebibytes(5),
retries: 5,
});
declare const bucket: s3.Bucket;
const s3Destination = new firehose.S3Bucket(bucket, {
processors: [lambdaProcessor],
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
destination: s3Destination,
});
python
# bucket: s3.Bucket
# Provide a Lambda function that will transform records before delivery, with custom
# buffering and retry configuration
lambda_function = lambda_.Function(self, "Processor",
runtime=lambda_.Runtime.NODEJS_LATEST,
handler="index.handler",
code=lambda_.Code.from_asset(path.join(__dirname, "process-records"))
)
lambda_processor = firehose.LambdaFunctionProcessor(lambda_function,
buffer_interval=Duration.minutes(5),
buffer_size=Size.mebibytes(5),
retries=5
)
s3_destination = firehose.S3Bucket(bucket,
processors=[lambda_processor]
)
firehose.DeliveryStream(self, "Delivery Stream",
destination=s3_destination
)
java
Bucket bucket;
// Provide a Lambda function that will transform records before delivery, with custom
// buffering and retry configuration
Function lambdaFunction = Function.Builder.create(this, "Processor")
.runtime(Runtime.NODEJS_LATEST)
.handler("index.handler")
.code(Code.fromAsset(join(__dirname, "process-records")))
.build();
LambdaFunctionProcessor lambdaProcessor = LambdaFunctionProcessor.Builder.create(lambdaFunction)
.bufferInterval(Duration.minutes(5))
.bufferSize(Size.mebibytes(5))
.retries(5)
.build();
S3Bucket s3Destination = S3Bucket.Builder.create(bucket)
.processors(List.of(lambdaProcessor))
.build();
DeliveryStream.Builder.create(this, "Delivery Stream")
.destination(s3Destination)
.build();
csharp
Bucket bucket;
// Provide a Lambda function that will transform records before delivery, with custom
// buffering and retry configuration
var lambdaFunction = new Function(this, "Processor", new FunctionProps {
Runtime = Runtime.NODEJS_LATEST,
Handler = "index.handler",
Code = Code.FromAsset(Join(__dirname, "process-records"))
});
var lambdaProcessor = new LambdaFunctionProcessor(lambdaFunction, new DataProcessorProps {
BufferInterval = Duration.Minutes(5),
BufferSize = Size.Mebibytes(5),
Retries = 5
});
var s3Destination = new S3Bucket(bucket, new S3BucketProps {
Processors = new [] { lambdaProcessor }
});
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Destination = s3Destination
});
go
var bucket Bucket
// Provide a Lambda function that will transform records before delivery, with custom
// buffering and retry configuration
lambdaFunction := lambda.NewFunction(this, jsii.String("Processor"), &FunctionProps{
Runtime: lambda.Runtime_NODEJS_LATEST(),
Handler: jsii.String("index.handler"),
Code: lambda.Code_FromAsset(path.join(__dirname, jsii.String("process-records"))),
})
lambdaProcessor := firehose.NewLambdaFunctionProcessor(lambdaFunction, &DataProcessorProps{
BufferInterval: awscdk.Duration_Minutes(jsii.Number(5)),
BufferSize: awscdk.Size_Mebibytes(jsii.Number(5)),
Retries: jsii.Number(5),
})
s3Destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
Processors: []IDataProcessor{
lambdaProcessor,
},
})
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Destination: s3Destination,
})
ts
#!/usr/bin/env node
import * as path from 'path';
import * as firehose from 'aws-cdk-lib/aws-kinesisfirehose';
import * as kms from 'aws-cdk-lib/aws-kms';
import * as lambdanodejs from 'aws-cdk-lib/aws-lambda-nodejs';
import * as logs from 'aws-cdk-lib/aws-logs';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as cdk from 'aws-cdk-lib';
import { AwsApiCall, ExpectedResult, IntegTest } from '@aws-cdk/integ-tests-alpha';
const app = new cdk.App({
postCliContext: {
'@aws-cdk/aws-lambda:useCdkManagedLogGroup': false,
},
});
const stack = new cdk.Stack(app, 'aws-cdk-firehose-delivery-stream-s3-all-properties');
const bucket = new s3.Bucket(stack, 'FirehoseDeliveryStreamS3AllPropertiesBucket', {
removalPolicy: cdk.RemovalPolicy.DESTROY,
autoDeleteObjects: true,
});
const backupBucket = new s3.Bucket(stack, 'FirehoseDeliveryStreamS3AllPropertiesBackupBucket', {
removalPolicy: cdk.RemovalPolicy.DESTROY,
autoDeleteObjects: true,
});
const logGroup = new logs.LogGroup(stack, 'LogGroup', {
removalPolicy: cdk.RemovalPolicy.DESTROY,
});
const dataProcessorFunction = new lambdanodejs.NodejsFunction(stack, 'DataProcessorFunction', {
entry: path.join(__dirname, 'lambda-data-processor.js'),
timeout: cdk.Duration.minutes(1),
});
const processor = new firehose.LambdaFunctionProcessor(dataProcessorFunction, {
bufferInterval: cdk.Duration.seconds(60),
bufferSize: cdk.Size.mebibytes(1),
retries: 1,
});
const key = new kms.Key(stack, 'Key', {
removalPolicy: cdk.RemovalPolicy.DESTROY,
});
const backupKey = new kms.Key(stack, 'BackupKey', {
removalPolicy: cdk.RemovalPolicy.DESTROY,
});
const deliveryStream = new firehose.DeliveryStream(stack, 'DeliveryStream', {
destination: new firehose.S3Bucket(bucket, {
loggingConfig: new firehose.EnableLogging(logGroup),
processors: [processor],
compression: firehose.Compression.GZIP,
dataOutputPrefix: 'regularPrefix',
errorOutputPrefix: 'errorPrefix',
fileExtension: '.log.gz',
timeZone: cdk.TimeZone.ASIA_TOKYO,
bufferingInterval: cdk.Duration.seconds(60),
bufferingSize: cdk.Size.mebibytes(1),
encryptionKey: key,
s3Backup: {
mode: firehose.BackupMode.ALL,
bucket: backupBucket,
compression: firehose.Compression.ZIP,
dataOutputPrefix: 'backupPrefix',
errorOutputPrefix: 'backupErrorPrefix',
bufferingInterval: cdk.Duration.seconds(60),
bufferingSize: cdk.Size.mebibytes(1),
encryptionKey: backupKey,
},
}),
});
new firehose.DeliveryStream(stack, 'ZeroBufferingDeliveryStream', {
destination: new firehose.S3Bucket(bucket, {
compression: firehose.Compression.GZIP,
dataOutputPrefix: 'regularPrefix',
errorOutputPrefix: 'errorPrefix',
bufferingInterval: cdk.Duration.seconds(0),
}),
});
const testCase = new IntegTest(app, 'integ-tests', {
testCases: [stack],
regions: ['us-east-1'],
});
testCase.assertions.awsApiCall('Firehose', 'putRecord', {
DeliveryStreamName: deliveryStream.deliveryStreamName,
Record: {
Data: 'testData123',
},
});
const s3ApiCall = testCase.assertions.awsApiCall('S3', 'listObjectsV2', {
Bucket: bucket.bucketName,
MaxKeys: 1,
}).expect(ExpectedResult.objectLike({
KeyCount: 1,
})).waitForAssertions({
interval: cdk.Duration.seconds(30),
totalTimeout: cdk.Duration.minutes(10),
});
if (s3ApiCall instanceof AwsApiCall && s3ApiCall.waiterProvider) {
s3ApiCall.waiterProvider.addToRolePolicy({
Effect: 'Allow',
Action: ['s3:GetObject', 's3:ListBucket'],
Resource: ['*'],
});
}
python
import path as path
import aws_cdk.aws_kinesisfirehose as firehose
import aws_cdk.aws_kms as kms
import aws_cdk.aws_lambda_nodejs as lambdanodejs
import aws_cdk.aws_logs as logs
import aws_cdk.aws_s3 as s3
import aws_cdk as cdk
from aws_cdk.integ_tests_alpha import AwsApiCall, ExpectedResult, IntegTest
app = cdk.App(
post_cli_context={
"@aws-cdk/aws-lambda:useCdkManagedLogGroup": False
}
)
stack = cdk.Stack(app, "aws-cdk-firehose-delivery-stream-s3-all-properties")
bucket = s3.Bucket(stack, "FirehoseDeliveryStreamS3AllPropertiesBucket",
removal_policy=cdk.RemovalPolicy.DESTROY,
auto_delete_objects=True
)
backup_bucket = s3.Bucket(stack, "FirehoseDeliveryStreamS3AllPropertiesBackupBucket",
removal_policy=cdk.RemovalPolicy.DESTROY,
auto_delete_objects=True
)
log_group = logs.LogGroup(stack, "LogGroup",
removal_policy=cdk.RemovalPolicy.DESTROY
)
data_processor_function = lambdanodejs.NodejsFunction(stack, "DataProcessorFunction",
entry=path.join(__dirname, "lambda-data-processor.js"),
timeout=cdk.Duration.minutes(1)
)
processor = firehose.LambdaFunctionProcessor(data_processor_function,
buffer_interval=cdk.Duration.seconds(60),
buffer_size=cdk.Size.mebibytes(1),
retries=1
)
key = kms.Key(stack, "Key",
removal_policy=cdk.RemovalPolicy.DESTROY
)
backup_key = kms.Key(stack, "BackupKey",
removal_policy=cdk.RemovalPolicy.DESTROY
)
delivery_stream = firehose.DeliveryStream(stack, "DeliveryStream",
destination=firehose.S3Bucket(bucket,
logging_config=firehose.EnableLogging(log_group),
processors=[processor],
compression=firehose.Compression.GZIP,
data_output_prefix="regularPrefix",
error_output_prefix="errorPrefix",
file_extension=".log.gz",
time_zone=cdk.TimeZone.ASIA_TOKYO,
buffering_interval=cdk.Duration.seconds(60),
buffering_size=cdk.Size.mebibytes(1),
encryption_key=key,
s3_backup=firehose.DestinationS3BackupProps(
mode=firehose.BackupMode.ALL,
bucket=backup_bucket,
compression=firehose.Compression.ZIP,
data_output_prefix="backupPrefix",
error_output_prefix="backupErrorPrefix",
buffering_interval=cdk.Duration.seconds(60),
buffering_size=cdk.Size.mebibytes(1),
encryption_key=backup_key
)
)
)
firehose.DeliveryStream(stack, "ZeroBufferingDeliveryStream",
destination=firehose.S3Bucket(bucket,
compression=firehose.Compression.GZIP,
data_output_prefix="regularPrefix",
error_output_prefix="errorPrefix",
buffering_interval=cdk.Duration.seconds(0)
)
)
test_case = IntegTest(app, "integ-tests",
test_cases=[stack],
regions=["us-east-1"]
)
test_case.assertions.aws_api_call("Firehose", "putRecord", {
"DeliveryStreamName": delivery_stream.delivery_stream_name,
"Record": {
"Data": "testData123"
}
})
s3_api_call = test_case.assertions.aws_api_call("S3", "listObjectsV2", {
"Bucket": bucket.bucket_name,
"MaxKeys": 1
}).expect(ExpectedResult.object_like({
"KeyCount": 1
})).wait_for_assertions(
interval=cdk.Duration.seconds(30),
total_timeout=cdk.Duration.minutes(10)
)
if s3_api_call instanceof AwsApiCall && s3_api_call.waiter_provider:
s3_api_call.waiter_provider.add_to_role_policy({
"Effect": "Allow",
"Action": ["s3:GetObject", "s3:ListBucket"],
"Resource": ["*"]
})
java
import path.*;
import software.amazon.awscdk.services.kinesisfirehose.*;
import software.amazon.awscdk.services.kms.*;
import software.amazon.awscdk.services.lambda.nodejs.*;
import software.amazon.awscdk.services.logs.*;
import software.amazon.awscdk.services.s3.*;
import software.amazon.awscdk.*;
import software.amazon.awscdk.integtests.alpha.AwsApiCall;
import software.amazon.awscdk.integtests.alpha.ExpectedResult;
import software.amazon.awscdk.integtests.alpha.IntegTest;
App app = App.Builder.create()
.postCliContext(Map.of(
"@aws-cdk/aws-lambda:useCdkManagedLogGroup", false))
.build();
Stack stack = new Stack(app, "aws-cdk-firehose-delivery-stream-s3-all-properties");
Bucket bucket = Bucket.Builder.create(stack, "FirehoseDeliveryStreamS3AllPropertiesBucket")
.removalPolicy(RemovalPolicy.DESTROY)
.autoDeleteObjects(true)
.build();
Bucket backupBucket = Bucket.Builder.create(stack, "FirehoseDeliveryStreamS3AllPropertiesBackupBucket")
.removalPolicy(RemovalPolicy.DESTROY)
.autoDeleteObjects(true)
.build();
LogGroup logGroup = LogGroup.Builder.create(stack, "LogGroup")
.removalPolicy(RemovalPolicy.DESTROY)
.build();
NodejsFunction dataProcessorFunction = NodejsFunction.Builder.create(stack, "DataProcessorFunction")
.entry(join(__dirname, "lambda-data-processor.js"))
.timeout(Duration.minutes(1))
.build();
LambdaFunctionProcessor processor = LambdaFunctionProcessor.Builder.create(dataProcessorFunction)
.bufferInterval(Duration.seconds(60))
.bufferSize(Size.mebibytes(1))
.retries(1)
.build();
Key key = Key.Builder.create(stack, "Key")
.removalPolicy(RemovalPolicy.DESTROY)
.build();
Key backupKey = Key.Builder.create(stack, "BackupKey")
.removalPolicy(RemovalPolicy.DESTROY)
.build();
DeliveryStream deliveryStream = DeliveryStream.Builder.create(stack, "DeliveryStream")
.destination(S3Bucket.Builder.create(bucket)
.loggingConfig(new EnableLogging(logGroup))
.processors(List.of(processor))
.compression(Compression.GZIP)
.dataOutputPrefix("regularPrefix")
.errorOutputPrefix("errorPrefix")
.fileExtension(".log.gz")
.timeZone(TimeZone.ASIA_TOKYO)
.bufferingInterval(Duration.seconds(60))
.bufferingSize(Size.mebibytes(1))
.encryptionKey(key)
.s3Backup(DestinationS3BackupProps.builder()
.mode(BackupMode.ALL)
.bucket(backupBucket)
.compression(Compression.ZIP)
.dataOutputPrefix("backupPrefix")
.errorOutputPrefix("backupErrorPrefix")
.bufferingInterval(Duration.seconds(60))
.bufferingSize(Size.mebibytes(1))
.encryptionKey(backupKey)
.build())
.build())
.build();
DeliveryStream.Builder.create(stack, "ZeroBufferingDeliveryStream")
.destination(S3Bucket.Builder.create(bucket)
.compression(Compression.GZIP)
.dataOutputPrefix("regularPrefix")
.errorOutputPrefix("errorPrefix")
.bufferingInterval(Duration.seconds(0))
.build())
.build();
IntegTest testCase = IntegTest.Builder.create(app, "integ-tests")
.testCases(List.of(stack))
.regions(List.of("us-east-1"))
.build();
testCase.assertions.awsApiCall("Firehose", "putRecord", Map.of(
"DeliveryStreamName", deliveryStream.getDeliveryStreamName(),
"Record", Map.of(
"Data", "testData123")));
IApiCall s3ApiCall = testCase.assertions.awsApiCall("S3", "listObjectsV2", Map.of(
"Bucket", bucket.bucketName,
"MaxKeys", 1)).expect(ExpectedResult.objectLike(Map.of(
"KeyCount", 1))).waitForAssertions(WaiterStateMachineOptions.builder()
.interval(Duration.seconds(30))
.totalTimeout(Duration.minutes(10))
.build());
if (s3ApiCall instanceof AwsApiCall && s3ApiCall.getWaiterProvider()) {
s3ApiCall.waiterProvider.addToRolePolicy(Map.of(
"Effect", "Allow",
"Action", List.of("s3:GetObject", "s3:ListBucket"),
"Resource", List.of("*")));
}
csharp
using Path;
using Amazon.CDK.AWS.KinesisFirehose;
using Amazon.CDK.AWS.KMS;
using Amazon.CDK.AWS.Lambda.Nodejs;
using Amazon.CDK.AWS.Logs;
using Amazon.CDK.AWS.S3;
using Amazon.CDK;
using Amazon.CDK.IntegTests.Alpha;
var app = new App(new AppProps {
PostCliContext = new Dictionary<string, object> {
{ "@aws-cdk/aws-lambda:useCdkManagedLogGroup", false }
}
});
var stack = new Stack(app, "aws-cdk-firehose-delivery-stream-s3-all-properties");
var bucket = new Bucket(stack, "FirehoseDeliveryStreamS3AllPropertiesBucket", new BucketProps {
RemovalPolicy = RemovalPolicy.DESTROY,
AutoDeleteObjects = true
});
var backupBucket = new Bucket(stack, "FirehoseDeliveryStreamS3AllPropertiesBackupBucket", new BucketProps {
RemovalPolicy = RemovalPolicy.DESTROY,
AutoDeleteObjects = true
});
var logGroup = new LogGroup(stack, "LogGroup", new LogGroupProps {
RemovalPolicy = RemovalPolicy.DESTROY
});
var dataProcessorFunction = new NodejsFunction(stack, "DataProcessorFunction", new NodejsFunctionProps {
Entry = Join(__dirname, "lambda-data-processor.js"),
Timeout = Duration.Minutes(1)
});
var processor = new LambdaFunctionProcessor(dataProcessorFunction, new DataProcessorProps {
BufferInterval = Duration.Seconds(60),
BufferSize = Size.Mebibytes(1),
Retries = 1
});
var key = new Key(stack, "Key", new KeyProps {
RemovalPolicy = RemovalPolicy.DESTROY
});
var backupKey = new Key(stack, "BackupKey", new KeyProps {
RemovalPolicy = RemovalPolicy.DESTROY
});
var deliveryStream = new DeliveryStream(stack, "DeliveryStream", new DeliveryStreamProps {
Destination = new S3Bucket(bucket, new S3BucketProps {
LoggingConfig = new EnableLogging(logGroup),
Processors = new [] { processor },
Compression = Compression.GZIP,
DataOutputPrefix = "regularPrefix",
ErrorOutputPrefix = "errorPrefix",
FileExtension = ".log.gz",
TimeZone = TimeZone.ASIA_TOKYO,
BufferingInterval = Duration.Seconds(60),
BufferingSize = Size.Mebibytes(1),
EncryptionKey = key,
S3Backup = new DestinationS3BackupProps {
Mode = BackupMode.ALL,
Bucket = backupBucket,
Compression = Compression.ZIP,
DataOutputPrefix = "backupPrefix",
ErrorOutputPrefix = "backupErrorPrefix",
BufferingInterval = Duration.Seconds(60),
BufferingSize = Size.Mebibytes(1),
EncryptionKey = backupKey
}
})
});
new DeliveryStream(stack, "ZeroBufferingDeliveryStream", new DeliveryStreamProps {
Destination = new S3Bucket(bucket, new S3BucketProps {
Compression = Compression.GZIP,
DataOutputPrefix = "regularPrefix",
ErrorOutputPrefix = "errorPrefix",
BufferingInterval = Duration.Seconds(0)
})
});
var testCase = new IntegTest(app, "integ-tests", new IntegTestProps {
TestCases = new [] { stack },
Regions = new [] { "us-east-1" }
});
testCase.Assertions.AwsApiCall("Firehose", "putRecord", new Dictionary<string, object> {
{ "DeliveryStreamName", deliveryStream.DeliveryStreamName },
{ "Record", new Struct {
Data = "testData123"
} }
});
var s3ApiCall = testCase.Assertions.AwsApiCall("S3", "listObjectsV2", new Dictionary<string, object> {
{ "Bucket", bucket.BucketName },
{ "MaxKeys", 1 }
}).Expect(ExpectedResult.ObjectLike(new Dictionary<string, object> {
{ "KeyCount", 1 }
})).WaitForAssertions(new WaiterStateMachineOptions {
Interval = Duration.Seconds(30),
TotalTimeout = Duration.Minutes(10)
});
if (s3ApiCall instanceof AwsApiCall && s3ApiCall.WaiterProvider)
{
s3ApiCall.WaiterProvider.AddToRolePolicy(new Dictionary<string, object> {
{ "Effect", "Allow" },
{ "Action", new [] { "s3:GetObject", "s3:ListBucket" } },
{ "Resource", new [] { "*" } }
});
}
go
import path "github.com/aws-samples/dummy/path"
import "github.com/aws/aws-cdk-go/awscdk"
import "github.com/aws/aws-cdk-go/awscdk"
import lambdanodejs "github.com/aws/aws-cdk-go/awscdk"
import logs "github.com/aws/aws-cdk-go/awscdk"
import "github.com/aws/aws-cdk-go/awscdk"
import "github.com/aws/aws-cdk-go/awscdk"
import "github.com/aws/aws-cdk-go/awscdkintegtestsalpha"
app := cdk.NewApp(&AppProps{
PostCliContext: map[string]interface{}{
"@aws-cdk/aws-lambda:useCdkManagedLogGroup": jsii.Boolean(false),
},
})
stack := cdk.NewStack(app, jsii.String("aws-cdk-firehose-delivery-stream-s3-all-properties"))
bucket := s3.NewBucket(stack, jsii.String("FirehoseDeliveryStreamS3AllPropertiesBucket"), &BucketProps{
RemovalPolicy: cdk.RemovalPolicy_DESTROY,
AutoDeleteObjects: jsii.Boolean(true),
})
backupBucket := s3.NewBucket(stack, jsii.String("FirehoseDeliveryStreamS3AllPropertiesBackupBucket"), &BucketProps{
RemovalPolicy: cdk.RemovalPolicy_DESTROY,
AutoDeleteObjects: jsii.Boolean(true),
})
logGroup := logs.NewLogGroup(stack, jsii.String("LogGroup"), &LogGroupProps{
RemovalPolicy: cdk.RemovalPolicy_DESTROY,
})
dataProcessorFunction := lambdanodejs.NewNodejsFunction(stack, jsii.String("DataProcessorFunction"), &NodejsFunctionProps{
Entry: path.join(__dirname, jsii.String("lambda-data-processor.js")),
Timeout: cdk.Duration_Minutes(jsii.Number(1)),
})
processor := firehose.NewLambdaFunctionProcessor(dataProcessorFunction, &DataProcessorProps{
BufferInterval: cdk.Duration_Seconds(jsii.Number(60)),
BufferSize: cdk.Size_Mebibytes(jsii.Number(1)),
Retries: jsii.Number(1),
})
key := kms.NewKey(stack, jsii.String("Key"), &KeyProps{
RemovalPolicy: cdk.RemovalPolicy_DESTROY,
})
backupKey := kms.NewKey(stack, jsii.String("BackupKey"), &KeyProps{
RemovalPolicy: cdk.RemovalPolicy_DESTROY,
})
deliveryStream := firehose.NewDeliveryStream(stack, jsii.String("DeliveryStream"), &DeliveryStreamProps{
Destination: firehose.NewS3Bucket(bucket, &S3BucketProps{
LoggingConfig: firehose.NewEnableLogging(logGroup),
Processors: []IDataProcessor{
processor,
},
Compression: firehose.Compression_GZIP(),
DataOutputPrefix: jsii.String("regularPrefix"),
ErrorOutputPrefix: jsii.String("errorPrefix"),
FileExtension: jsii.String(".log.gz"),
TimeZone: cdk.TimeZone_ASIA_TOKYO(),
BufferingInterval: cdk.Duration_*Seconds(jsii.Number(60)),
BufferingSize: cdk.Size_*Mebibytes(jsii.Number(1)),
EncryptionKey: key,
S3Backup: &DestinationS3BackupProps{
Mode: firehose.BackupMode_ALL,
Bucket: backupBucket,
Compression: firehose.Compression_ZIP(),
DataOutputPrefix: jsii.String("backupPrefix"),
ErrorOutputPrefix: jsii.String("backupErrorPrefix"),
BufferingInterval: cdk.Duration_*Seconds(jsii.Number(60)),
BufferingSize: cdk.Size_*Mebibytes(jsii.Number(1)),
EncryptionKey: backupKey,
},
}),
})
firehose.NewDeliveryStream(stack, jsii.String("ZeroBufferingDeliveryStream"), &DeliveryStreamProps{
Destination: firehose.NewS3Bucket(bucket, &S3BucketProps{
Compression: firehose.Compression_GZIP(),
DataOutputPrefix: jsii.String("regularPrefix"),
ErrorOutputPrefix: jsii.String("errorPrefix"),
BufferingInterval: cdk.Duration_*Seconds(jsii.Number(0)),
}),
})
testCase := awscdkintegtestsalpha.NewIntegTest(app, jsii.String("integ-tests"), &IntegTestProps{
TestCases: []Stack{
stack,
},
Regions: []*string{
jsii.String("us-east-1"),
},
})
testCase.Assertions.AwsApiCall(jsii.String("Firehose"), jsii.String("putRecord"), map[string]interface{}{
"DeliveryStreamName": deliveryStream.deliveryStreamName,
"Record": map[string]*string{
"Data": jsii.String("testData123"),
},
})
s3ApiCall := testCase.Assertions.AwsApiCall(jsii.String("S3"), jsii.String("listObjectsV2"), map[string]interface{}{
"Bucket": bucket.bucketName,
"MaxKeys": jsii.Number(1),
}).Expect(awscdkintegtestsalpha.ExpectedResult_ObjectLike(map[string]interface{}{
"KeyCount": jsii.Number(1),
})).WaitForAssertions(&WaiterStateMachineOptions{
Interval: cdk.Duration_*Seconds(jsii.Number(30)),
TotalTimeout: cdk.Duration_*Minutes(jsii.Number(10)),
})
if s3ApiCall instanceof awscdkintegtestsalpha.AwsApiCall && s3ApiCall.WaiterProvider {
s3ApiCall.WaiterProvider.AddToRolePolicy(map[string]interface{}{
"Effect": jsii.String("Allow"),
"Action": []*string{
jsii.String("s3:GetObject"),
jsii.String("s3:ListBucket"),
},
"Resource": []*string{
jsii.String("*"),
},
})
}
See: Data Transformation in the Amazon Data Firehose Developer Guide.
Add a new line delimiter when delivering data to Amazon S3
You can specify the AppendDelimiterToRecordProcessor built-in processor to add a new line delimiter between records in objects that are delivered to Amazon S3. This can be helpful for parsing objects in Amazon S3. For details, see Use Amazon S3 bucket prefix to deliver data.
ts
declare const bucket: s3.Bucket;
const s3Destination = new firehose.S3Bucket(bucket, {
processors: [
new firehose.AppendDelimiterToRecordProcessor(),
],
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
destination: s3Destination,
});
python
# bucket: s3.Bucket
s3_destination = firehose.S3Bucket(bucket,
processors=[
firehose.AppendDelimiterToRecordProcessor()
]
)
firehose.DeliveryStream(self, "Delivery Stream",
destination=s3_destination
)
java
Bucket bucket;
S3Bucket s3Destination = S3Bucket.Builder.create(bucket)
.processors(List.of(
new AppendDelimiterToRecordProcessor()))
.build();
DeliveryStream.Builder.create(this, "Delivery Stream")
.destination(s3Destination)
.build();
csharp
Bucket bucket;
var s3Destination = new S3Bucket(bucket, new S3BucketProps {
Processors = new [] {
new AppendDelimiterToRecordProcessor() }
});
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Destination = s3Destination
});
go
var bucket Bucket
s3Destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
Processors: []IDataProcessor{
firehose.NewAppendDelimiterToRecordProcessor(),
},
})
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Destination: s3Destination,
})
Decompress and extract message of CloudWatch Logs
CloudWatch Logs events are sent to Firehose in compressed gzip format. If you want to deliver decompressed log events to Firehose destinations, you can use the DecompressionProcessor to automatically decompress CloudWatch Logs. For details, see Send CloudWatch Logs to Firehose.
You may also needed to specify AppendDelimiterToRecordProcessor because decompressed log events record has no trailing newline.
ts
declare const bucket: s3.Bucket;
const s3Destination = new firehose.S3Bucket(bucket, {
processors: [
new firehose.DecompressionProcessor(),
new firehose.AppendDelimiterToRecordProcessor(),
],
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
destination: s3Destination,
});
python
# bucket: s3.Bucket
s3_destination = firehose.S3Bucket(bucket,
processors=[
firehose.DecompressionProcessor(),
firehose.AppendDelimiterToRecordProcessor()
]
)
firehose.DeliveryStream(self, "Delivery Stream",
destination=s3_destination
)
java
Bucket bucket;
S3Bucket s3Destination = S3Bucket.Builder.create(bucket)
.processors(List.of(
new DecompressionProcessor(),
new AppendDelimiterToRecordProcessor()))
.build();
DeliveryStream.Builder.create(this, "Delivery Stream")
.destination(s3Destination)
.build();
csharp
Bucket bucket;
var s3Destination = new S3Bucket(bucket, new S3BucketProps {
Processors = new [] {
new DecompressionProcessor(),
new AppendDelimiterToRecordProcessor() }
});
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Destination = s3Destination
});
go
var bucket Bucket
s3Destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
Processors: []IDataProcessor{
firehose.NewDecompressionProcessor(),
firehose.NewAppendDelimiterToRecordProcessor(),
},
})
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Destination: s3Destination,
})
When you enable decompression, you have the option to also enable message extraction. When using message extraction, Firehose filters out all metadata, such as owner, loggroup, logstream, and others from the decompressed CloudWatch Logs records and delivers only the content inside the message fields.
ts
declare const bucket: s3.Bucket;
const s3Destination = new firehose.S3Bucket(bucket, {
processors: [
new firehose.DecompressionProcessor(),
new firehose.CloudWatchLogProcessor({ dataMessageExtraction: true }),
],
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
destination: s3Destination,
});
python
# bucket: s3.Bucket
s3_destination = firehose.S3Bucket(bucket,
processors=[
firehose.DecompressionProcessor(),
firehose.CloudWatchLogProcessor(data_message_extraction=True)
]
)
firehose.DeliveryStream(self, "Delivery Stream",
destination=s3_destination
)
java
Bucket bucket;
S3Bucket s3Destination = S3Bucket.Builder.create(bucket)
.processors(List.of(
new DecompressionProcessor(),
CloudWatchLogProcessor.Builder.create().dataMessageExtraction(true).build()))
.build();
DeliveryStream.Builder.create(this, "Delivery Stream")
.destination(s3Destination)
.build();
csharp
Bucket bucket;
var s3Destination = new S3Bucket(bucket, new S3BucketProps {
Processors = new [] {
new DecompressionProcessor(),
new CloudWatchLogProcessor(new CloudWatchLogProcessorOptions { DataMessageExtraction = true }) }
});
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Destination = s3Destination
});
go
var bucket Bucket
s3Destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
Processors: []IDataProcessor{
firehose.NewDecompressionProcessor(),
firehose.NewCloudWatchLogProcessor(&CloudWatchLogProcessorOptions{
DataMessageExtraction: jsii.Boolean(true),
}),
},
})
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Destination: s3Destination,
})
Dynamic Partitioning
Dynamic partitioning enables you to continuously partition streaming data in Firehose by using keys within data (for example, customer_id or transaction_id) and then deliver the data grouped by these keys into corresponding Amazon S3 prefixes.
For details, see Partition streaming data in Amazon Data Firehose.
Create partitioning keys with inline parsing
To enable dynamic partitioning with inline parsing, specify MetadataExtractionProcessor data processor in processors.
Only supported mechanism currently is jq 1.6 parser.
The partition keys can be referred as !{partitionKeyFromQuery:key} in dataOutputPrefix.
ts
declare const bucket: s3.Bucket;
const s3Destination = new firehose.S3Bucket(bucket, {
dynamicPartitioning: { enabled: true },
processors: [
firehose.MetadataExtractionProcessor.jq16({
customer_id: '.customer_id',
device: '.type.device',
year: '.event_timestamp|strftime("%Y")',
}),
],
dataOutputPrefix: '!{partitionKeyFromQuery:year}/!{partitionKeyFromQuery:device}/!{partitionKeyFromQuery:customer_id}/',
});
new firehose.DeliveryStream(this, 'DeliveryStream', {
destination: s3Destination,
});
python
# bucket: s3.Bucket
s3_destination = firehose.S3Bucket(bucket,
dynamic_partitioning=firehose.DynamicPartitioningProps(enabled=True),
processors=[
firehose.MetadataExtractionProcessor.jq16({
"customer_id": ".customer_id",
"device": ".type.device",
"year": ".event_timestamp|strftime(\"%Y\")"
})
],
data_output_prefix="!{partitionKeyFromQuery:year}/!{partitionKeyFromQuery:device}/!{partitionKeyFromQuery:customer_id}/"
)
firehose.DeliveryStream(self, "DeliveryStream",
destination=s3_destination
)
java
Bucket bucket;
S3Bucket s3Destination = S3Bucket.Builder.create(bucket)
.dynamicPartitioning(DynamicPartitioningProps.builder().enabled(true).build())
.processors(List.of(MetadataExtractionProcessor.jq16(Map.of(
"customer_id", ".customer_id",
"device", ".type.device",
"year", ".event_timestamp|strftime(\"%Y\")"))))
.dataOutputPrefix("!{partitionKeyFromQuery:year}/!{partitionKeyFromQuery:device}/!{partitionKeyFromQuery:customer_id}/")
.build();
DeliveryStream.Builder.create(this, "DeliveryStream")
.destination(s3Destination)
.build();
csharp
Bucket bucket;
var s3Destination = new S3Bucket(bucket, new S3BucketProps {
DynamicPartitioning = new DynamicPartitioningProps { Enabled = true },
Processors = new [] { MetadataExtractionProcessor.Jq16(new Dictionary<string, string> {
{ "customer_id", ".customer_id" },
{ "device", ".type.device" },
{ "year", ".event_timestamp|strftime(\"%Y\")" }
}) },
DataOutputPrefix = "!{partitionKeyFromQuery:year}/!{partitionKeyFromQuery:device}/!{partitionKeyFromQuery:customer_id}/"
});
new DeliveryStream(this, "DeliveryStream", new DeliveryStreamProps {
Destination = s3Destination
});
go
var bucket Bucket
s3Destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
DynamicPartitioning: &DynamicPartitioningProps{
Enabled: jsii.Boolean(true),
},
Processors: []IDataProcessor{
firehose.MetadataExtractionProcessor_Jq16(map[string]*string{
"customer_id": jsii.String(".customer_id"),
"device": jsii.String(".type.device"),
"year": jsii.String(".event_timestamp|strftime(\"%Y\")"),
}),
},
DataOutputPrefix: jsii.String("!{partitionKeyFromQuery:year}/!{partitionKeyFromQuery:device}/!{partitionKeyFromQuery:customer_id}/"),
})
firehose.NewDeliveryStream(this, jsii.String("DeliveryStream"), &DeliveryStreamProps{
Destination: s3Destination,
})
Create partitioning keys with an AWS Lambda function
For compressed or encrypted data records, or data that is in any file format other than JSON, you can use the LambdaFunctionDataProcessor with your own custom code to decompress, decrypt, or transform the records in order to extract and return the data fields needed for partitioning.
The lambda function must return metadata in your result records.
The partition keys can be referred as !{partitionKeyFromLambda:key} in dataOutputPrefix.
ts
declare const bucket: s3.Bucket;
declare const lambdaFunction: lambda.Function;
const s3Destination = new firehose.S3Bucket(bucket, {
dynamicPartitioning: { enabled: true },
processors: [
new firehose.LambdaFunctionProcessor(lambdaFunction),
],
dataOutputPrefix: '!{partitionKeyFromLambda:year}/!{partitionKeyFromLambda:device}/!{partitionKeyFromLambda:customer_id}/',
});
new firehose.DeliveryStream(this, 'DeliveryStream', {
destination: s3Destination,
});
python
# bucket: s3.Bucket
# lambda_function: lambda.Function
s3_destination = firehose.S3Bucket(bucket,
dynamic_partitioning=firehose.DynamicPartitioningProps(enabled=True),
processors=[
firehose.LambdaFunctionProcessor(lambda_function)
],
data_output_prefix="!{partitionKeyFromLambda:year}/!{partitionKeyFromLambda:device}/!{partitionKeyFromLambda:customer_id}/"
)
firehose.DeliveryStream(self, "DeliveryStream",
destination=s3_destination
)
java
Bucket bucket;
Function lambdaFunction;
S3Bucket s3Destination = S3Bucket.Builder.create(bucket)
.dynamicPartitioning(DynamicPartitioningProps.builder().enabled(true).build())
.processors(List.of(
new LambdaFunctionProcessor(lambdaFunction)))
.dataOutputPrefix("!{partitionKeyFromLambda:year}/!{partitionKeyFromLambda:device}/!{partitionKeyFromLambda:customer_id}/")
.build();
DeliveryStream.Builder.create(this, "DeliveryStream")
.destination(s3Destination)
.build();
csharp
Bucket bucket;
Function lambdaFunction;
var s3Destination = new S3Bucket(bucket, new S3BucketProps {
DynamicPartitioning = new DynamicPartitioningProps { Enabled = true },
Processors = new [] {
new LambdaFunctionProcessor(lambdaFunction) },
DataOutputPrefix = "!{partitionKeyFromLambda:year}/!{partitionKeyFromLambda:device}/!{partitionKeyFromLambda:customer_id}/"
});
new DeliveryStream(this, "DeliveryStream", new DeliveryStreamProps {
Destination = s3Destination
});
go
var bucket Bucket
var lambdaFunction Function
s3Destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
DynamicPartitioning: &DynamicPartitioningProps{
Enabled: jsii.Boolean(true),
},
Processors: []IDataProcessor{
firehose.NewLambdaFunctionProcessor(lambdaFunction),
},
DataOutputPrefix: jsii.String("!{partitionKeyFromLambda:year}/!{partitionKeyFromLambda:device}/!{partitionKeyFromLambda:customer_id}/"),
})
firehose.NewDeliveryStream(this, jsii.String("DeliveryStream"), &DeliveryStreamProps{
Destination: s3Destination,
})
Multi record deaggregation
To apply dynamic partitioning to aggregated data (for example, multiple events, logs, or records aggregated into a single PutRecord and PutRecordBatch API call), use RecordDeAggregationProcessor.
When the input data is JSON objects on a single line with no delimiter or newline-delimited (JSONL), specify RecordDeAggregationProcessor.json().
ts
declare const bucket: s3.Bucket;
const s3Destination = new firehose.S3Bucket(bucket, {
dynamicPartitioning: { enabled: true },
processors: [
firehose.RecordDeAggregationProcessor.json(),
],
});
python
# bucket: s3.Bucket
s3_destination = firehose.S3Bucket(bucket,
dynamic_partitioning=firehose.DynamicPartitioningProps(enabled=True),
processors=[
firehose.RecordDeAggregationProcessor.json()
]
)
java
Bucket bucket;
S3Bucket s3Destination = S3Bucket.Builder.create(bucket)
.dynamicPartitioning(DynamicPartitioningProps.builder().enabled(true).build())
.processors(List.of(RecordDeAggregationProcessor.json()))
.build();
csharp
Bucket bucket;
var s3Destination = new S3Bucket(bucket, new S3BucketProps {
DynamicPartitioning = new DynamicPartitioningProps { Enabled = true },
Processors = new [] { RecordDeAggregationProcessor.Json() }
});
go
var bucket Bucket
s3Destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
DynamicPartitioning: &DynamicPartitioningProps{
Enabled: jsii.Boolean(true),
},
Processors: []IDataProcessor{
firehose.RecordDeAggregationProcessor_Json(),
},
})
You can also specify custom delimiter using RecordDeAggregationProcessor.delimited().
ts
declare const bucket: s3.Bucket;
const s3Destination = new firehose.S3Bucket(bucket, {
dynamicPartitioning: { enabled: true },
processors: [
firehose.RecordDeAggregationProcessor.delimited('####'),
],
});
python
# bucket: s3.Bucket
s3_destination = firehose.S3Bucket(bucket,
dynamic_partitioning=firehose.DynamicPartitioningProps(enabled=True),
processors=[
firehose.RecordDeAggregationProcessor.delimited("####")
]
)
java
Bucket bucket;
S3Bucket s3Destination = S3Bucket.Builder.create(bucket)
.dynamicPartitioning(DynamicPartitioningProps.builder().enabled(true).build())
.processors(List.of(RecordDeAggregationProcessor.delimited("####")))
.build();
csharp
Bucket bucket;
var s3Destination = new S3Bucket(bucket, new S3BucketProps {
DynamicPartitioning = new DynamicPartitioningProps { Enabled = true },
Processors = new [] { RecordDeAggregationProcessor.Delimited("####") }
});
go
var bucket Bucket
s3Destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
DynamicPartitioning: &DynamicPartitioningProps{
Enabled: jsii.Boolean(true),
},
Processors: []IDataProcessor{
firehose.RecordDeAggregationProcessor_Delimited(jsii.String("####")),
},
})
Record deaggregation by JSON or by delimiter is capped at 500 per record.
Specifying an IAM role
The DeliveryStream class automatically creates IAM service roles with all the minimum necessary permissions for Amazon Data Firehose to access the resources referenced by your delivery stream. One service role is created for the delivery stream that allows Amazon Data Firehose to read from a Kinesis data stream (if one is configured as the delivery stream source) and for server-side encryption. Note that if the DeliveryStream is created without specifying a source or encryptionKey, this role is not created as it is not needed.
Another service role is created for each destination, which gives Amazon Data Firehose write access to the destination resource, as well as the ability to invoke data transformers and read schemas for record format conversion. If you wish, you may specify your own IAM role for either the delivery stream or the destination service role, or both. It must have the correct trust policy (it must allow Amazon Data Firehose to assume it) or delivery stream creation or data delivery will fail. Other required permissions to destination resources, encryption keys, etc., will be provided automatically.
ts
// Create service roles for the delivery stream and destination.
// These can be used for other purposes and granted access to different resources.
// They must include the Amazon Data Firehose service principal in their trust policies.
// Two separate roles are shown below, but the same role can be used for both purposes.
const deliveryStreamRole = new iam.Role(this, 'Delivery Stream Role', {
assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
});
const destinationRole = new iam.Role(this, 'Destination Role', {
assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
});
// Specify the roles created above when defining the destination and delivery stream.
declare const bucket: s3.Bucket;
const destination = new firehose.S3Bucket(bucket, { role: destinationRole });
new firehose.DeliveryStream(this, 'Delivery Stream', {
destination: destination,
role: deliveryStreamRole,
});
python
# Specify the roles created above when defining the destination and delivery stream.
# bucket: s3.Bucket
# Create service roles for the delivery stream and destination.
# These can be used for other purposes and granted access to different resources.
# They must include the Amazon Data Firehose service principal in their trust policies.
# Two separate roles are shown below, but the same role can be used for both purposes.
delivery_stream_role = iam.Role(self, "Delivery Stream Role",
assumed_by=iam.ServicePrincipal("firehose.amazonaws.com")
)
destination_role = iam.Role(self, "Destination Role",
assumed_by=iam.ServicePrincipal("firehose.amazonaws.com")
)
destination = firehose.S3Bucket(bucket, role=destination_role)
firehose.DeliveryStream(self, "Delivery Stream",
destination=destination,
role=delivery_stream_role
)
java
// Specify the roles created above when defining the destination and delivery stream.
Bucket bucket;
// Create service roles for the delivery stream and destination.
// These can be used for other purposes and granted access to different resources.
// They must include the Amazon Data Firehose service principal in their trust policies.
// Two separate roles are shown below, but the same role can be used for both purposes.
Role deliveryStreamRole = Role.Builder.create(this, "Delivery Stream Role")
.assumedBy(new ServicePrincipal("firehose.amazonaws.com"))
.build();
Role destinationRole = Role.Builder.create(this, "Destination Role")
.assumedBy(new ServicePrincipal("firehose.amazonaws.com"))
.build();
S3Bucket destination = S3Bucket.Builder.create(bucket).role(destinationRole).build();
DeliveryStream.Builder.create(this, "Delivery Stream")
.destination(destination)
.role(deliveryStreamRole)
.build();
csharp
// Specify the roles created above when defining the destination and delivery stream.
Bucket bucket;
// Create service roles for the delivery stream and destination.
// These can be used for other purposes and granted access to different resources.
// They must include the Amazon Data Firehose service principal in their trust policies.
// Two separate roles are shown below, but the same role can be used for both purposes.
var deliveryStreamRole = new Role(this, "Delivery Stream Role", new RoleProps {
AssumedBy = new ServicePrincipal("firehose.amazonaws.com")
});
var destinationRole = new Role(this, "Destination Role", new RoleProps {
AssumedBy = new ServicePrincipal("firehose.amazonaws.com")
});
var destination = new S3Bucket(bucket, new S3BucketProps { Role = destinationRole });
new DeliveryStream(this, "Delivery Stream", new DeliveryStreamProps {
Destination = destination,
Role = deliveryStreamRole
});
go
// Specify the roles created above when defining the destination and delivery stream.
var bucket Bucket
// Create service roles for the delivery stream and destination.
// These can be used for other purposes and granted access to different resources.
// They must include the Amazon Data Firehose service principal in their trust policies.
// Two separate roles are shown below, but the same role can be used for both purposes.
deliveryStreamRole := iam.NewRole(this, jsii.String("Delivery Stream Role"), &RoleProps{
AssumedBy: iam.NewServicePrincipal(jsii.String("firehose.amazonaws.com")),
})
destinationRole := iam.NewRole(this, jsii.String("Destination Role"), &RoleProps{
AssumedBy: iam.NewServicePrincipal(jsii.String("firehose.amazonaws.com")),
})
destination := firehose.NewS3Bucket(bucket, &S3BucketProps{
Role: destinationRole,
})
firehose.NewDeliveryStream(this, jsii.String("Delivery Stream"), &DeliveryStreamProps{
Destination: destination,
Role: deliveryStreamRole,
})
See Controlling Access in the Amazon Data Firehose Developer Guide.
Granting application access to a delivery stream
IAM roles, users or groups which need to be able to work with delivery streams should be granted IAM permissions.
Any object that implements the IGrantable interface (i.e., has an associated principal) can be granted permissions to a delivery stream by calling:
grantPutRecords(principal)- grants the principal the ability to put records onto the delivery streamgrant(principal, ...actions)- grants the principal permission to a custom set of actions
ts
const lambdaRole = new iam.Role(this, 'Role', {
assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
});
// Give the role permissions to write data to the delivery stream
declare const deliveryStream: firehose.DeliveryStream;
deliveryStream.grantPutRecords(lambdaRole);
python
# Give the role permissions to write data to the delivery stream
# delivery_stream: firehose.DeliveryStream
lambda_role = iam.Role(self, "Role",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
)
delivery_stream.grant_put_records(lambda_role)
java
// Give the role permissions to write data to the delivery stream
DeliveryStream deliveryStream;
Role lambdaRole = Role.Builder.create(this, "Role")
.assumedBy(new ServicePrincipal("lambda.amazonaws.com"))
.build();
deliveryStream.grantPutRecords(lambdaRole);
csharp
// Give the role permissions to write data to the delivery stream
DeliveryStream deliveryStream;
var lambdaRole = new Role(this, "Role", new RoleProps {
AssumedBy = new ServicePrincipal("lambda.amazonaws.com")
});
deliveryStream.GrantPutRecords(lambdaRole);
go
// Give the role permissions to write data to the delivery stream
var deliveryStream DeliveryStream
lambdaRole := iam.NewRole(this, jsii.String("Role"), &RoleProps{
AssumedBy: iam.NewServicePrincipal(jsii.String("lambda.amazonaws.com")),
})
deliveryStream.grantPutRecords(lambdaRole)
The following write permissions are provided to a service principal by the grantPutRecords() method:
firehose:PutRecordfirehose:PutRecordBatch
Granting a delivery stream access to a resource
Conversely to the above, Amazon Data Firehose requires permissions in order for delivery streams to interact with resources that you own. For example, if an S3 bucket is specified as a destination of a delivery stream, the delivery stream must be granted permissions to put and get objects from the bucket. When using the built-in AWS service destinations, the CDK grants the permissions automatically. However, custom or third-party destinations may require custom permissions. In this case, use the delivery stream as an IGrantable, as follows:
ts
const fn = new lambda.Function(this, 'Function', {
code: lambda.Code.fromInline('exports.handler = (event) => {}'),
runtime: lambda.Runtime.NODEJS_LATEST,
handler: 'index.handler',
});
declare const deliveryStream: firehose.DeliveryStream;
fn.grantInvoke(deliveryStream);
python
# delivery_stream: firehose.DeliveryStream
fn = lambda_.Function(self, "Function",
code=lambda_.Code.from_inline("exports.handler = (event) => {}"),
runtime=lambda_.Runtime.NODEJS_LATEST,
handler="index.handler"
)
fn.grant_invoke(delivery_stream)
java
DeliveryStream deliveryStream;
Function fn = Function.Builder.create(this, "Function")
.code(Code.fromInline("exports.handler = (event) => {}"))
.runtime(Runtime.NODEJS_LATEST)
.handler("index.handler")
.build();
fn.grantInvoke(deliveryStream);
csharp
DeliveryStream deliveryStream;
var fn = new Function(this, "Function", new FunctionProps {
Code = Code.FromInline("exports.handler = (event) => {}"),
Runtime = Runtime.NODEJS_LATEST,
Handler = "index.handler"
});
fn.GrantInvoke(deliveryStream);
go
var deliveryStream DeliveryStream
fn := lambda.NewFunction(this, jsii.String("Function"), &FunctionProps{
Code: lambda.Code_FromInline(jsii.String("exports.handler = (event) => {}")),
Runtime: lambda.Runtime_NODEJS_LATEST(),
Handler: jsii.String("index.handler"),
})
fn.GrantInvoke(deliveryStream)
Types
| Name | Kind | Summary |
|---|---|---|
DeliveryStream | Construct (L2) | Create a Amazon Data Firehose delivery stream. |
CfnDeliveryStream | Construct (L1) | The AWS::KinesisFirehose::DeliveryStream resource specifies 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. |
AppendDelimiterToRecordProcessor | Class | The data processor to append new line delimiter to each record. |
BackupMode | Enum | Options for S3 record backup of a delivery stream. |
CfnDeliveryStream.AmazonOpenSearchServerlessBufferingHintsProperty | Struct | Describes the buffering to perform before delivering data to the Serverless offering for Amazon OpenSearch Service destination. |
CfnDeliveryStream.AmazonOpenSearchServerlessDestinationConfigurationProperty | Struct | Describes the configuration of a destination in the Serverless offering for Amazon OpenSearch Service. |
CfnDeliveryStream.AmazonOpenSearchServerlessRetryOptionsProperty | Struct | Configures retry behavior in case Firehose is unable to deliver documents to the Serverless offering for Amazon OpenSearch Service. |
CfnDeliveryStream.AmazonopensearchserviceBufferingHintsProperty | Struct | Describes the buffering to perform before delivering data to the Amazon OpenSearch Service destination. |
CfnDeliveryStream.AmazonopensearchserviceDestinationConfigurationProperty | Struct | Describes the configuration of a destination in Amazon OpenSearch Service. |
CfnDeliveryStream.AmazonopensearchserviceRetryOptionsProperty | Struct | Configures retry behavior in case Kinesis Data Firehose is unable to deliver documents to Amazon OpenSearch Service. |
CfnDeliveryStream.AuthenticationConfigurationProperty | Struct | The authentication configuration of the Amazon MSK cluster. |
CfnDeliveryStream.BufferingHintsProperty | Struct | The BufferingHints property type specifies how Amazon Kinesis Data Firehose (Kinesis Data Firehose) buffers incoming data before delivering it to the destination. |
CfnDeliveryStream.CatalogConfigurationProperty | Struct | Describes the containers where the destination Apache Iceberg Tables are persisted. |
CfnDeliveryStream.CloudWatchLoggingOptionsProperty | Struct | The CloudWatchLoggingOptions property type specifies Amazon CloudWatch Logs (CloudWatch Logs) logging options that Amazon Kinesis Data Firehose (Kinesis Data Firehose) uses for the delivery stream. |
CfnDeliveryStream.CopyCommandProperty | Struct | The CopyCommand property type configures the Amazon Redshift COPY command that Amazon Kinesis Data Firehose (Kinesis Data Firehose) uses to load data into an Amazon Redshift cluster from an Amazon S3 bucket. |
CfnDeliveryStream.DatabaseColumnsProperty | Struct | |
CfnDeliveryStream.DatabaseSourceAuthenticationConfigurationProperty | Struct | The structure to configure the authentication methods for Firehose to connect to source database endpoint. |
CfnDeliveryStream.DatabaseSourceConfigurationProperty | Struct | The top level object for configuring streams with database as a source. |
CfnDeliveryStream.DatabaseSourceVPCConfigurationProperty | Struct | The structure for details of the VPC Endpoint Service which Firehose uses to create a PrivateLink to the database. |
CfnDeliveryStream.DatabasesProperty | Struct | |
CfnDeliveryStream.DatabaseTablesProperty | Struct | |
CfnDeliveryStream.DataFormatConversionConfigurationProperty | Struct | Specifies that you want Kinesis Data Firehose to convert data from the JSON format to the Parquet or ORC format before writing it to Amazon S3. |
CfnDeliveryStream.DeliveryStreamEncryptionConfigurationInputProperty | Struct | Specifies the type and Amazon Resource Name (ARN) of the CMK to use for Server-Side Encryption (SSE). |
CfnDeliveryStream.DeserializerProperty | Struct | The deserializer you want Kinesis Data Firehose to use for converting the input data from JSON. |
CfnDeliveryStream.DestinationTableConfigurationProperty | Struct | Describes the configuration of a destination in Apache Iceberg Tables. |
CfnDeliveryStream.DirectPutSourceConfigurationProperty | Struct | The structure that configures parameters such as ThroughputHintInMBs for a stream configured with Direct PUT as a source. |
CfnDeliveryStream.DocumentIdOptionsProperty | Struct | Indicates the method for setting up document ID. |
CfnDeliveryStream.DynamicPartitioningConfigurationProperty | Struct | The DynamicPartitioningConfiguration property type specifies the configuration of the dynamic partitioning mechanism that creates targeted data sets from the streaming data by partitioning it based on partition keys. |
CfnDeliveryStream.ElasticsearchBufferingHintsProperty | Struct | The ElasticsearchBufferingHints property type specifies how Amazon Kinesis Data Firehose (Kinesis Data Firehose) buffers incoming data while delivering it to the destination. |
CfnDeliveryStream.ElasticsearchDestinationConfigurationProperty | Struct | The ElasticsearchDestinationConfiguration property type specifies an Amazon Elasticsearch Service (Amazon ES) domain that Amazon Kinesis Data Firehose (Kinesis Data Firehose) delivers data to. |
CfnDeliveryStream.ElasticsearchRetryOptionsProperty | Struct | The ElasticsearchRetryOptions property type configures the retry behavior for when Amazon Kinesis Data Firehose (Kinesis Data Firehose) can't deliver data to Amazon Elasticsearch Service (Amazon ES). |
CfnDeliveryStream.EncryptionConfigurationProperty | Struct | The EncryptionConfiguration property type specifies the encryption settings that Amazon Kinesis Data Firehose (Kinesis Data Firehose) uses when delivering data to Amazon Simple Storage Service (Amazon S3). |
CfnDeliveryStream.ExtendedS3DestinationConfigurationProperty | Struct | The ExtendedS3DestinationConfiguration property type configures an Amazon S3 destination for an Amazon Kinesis Data Firehose delivery stream. |
CfnDeliveryStream.HiveJsonSerDeProperty | Struct | The native Hive / HCatalog JsonSerDe. |
CfnDeliveryStream.HttpEndpointCommonAttributeProperty | Struct | Describes the metadata that's delivered to the specified HTTP endpoint destination. |
CfnDeliveryStream.HttpEndpointConfigurationProperty | Struct | Describes the configuration of the HTTP endpoint to which Kinesis Firehose delivers data. |
CfnDeliveryStream.HttpEndpointDestinationConfigurationProperty | Struct | Describes the configuration of the HTTP endpoint destination. |
CfnDeliveryStream.HttpEndpointRequestConfigurationProperty | Struct | The configuration of the HTTP endpoint request. |
CfnDeliveryStream.IcebergDestinationConfigurationProperty | Struct | Specifies the destination configure settings for Apache Iceberg Table. |
CfnDeliveryStream.InputFormatConfigurationProperty | Struct | Specifies the deserializer you want to use to convert the format of the input data. |
CfnDeliveryStream.KinesisStreamSourceConfigurationProperty | Struct | The KinesisStreamSourceConfiguration property type specifies the stream and role Amazon Resource Names (ARNs) for a Kinesis stream used as the source for a delivery stream. |
CfnDeliveryStream.KMSEncryptionConfigProperty | Struct | The KMSEncryptionConfig property type specifies the AWS Key Management Service ( AWS KMS) encryption key that Amazon Simple Storage Service (Amazon S3) uses to encrypt data delivered by the Amazon Kinesis Data Firehose (Kinesis Data Firehose) stream. |
CfnDeliveryStream.MSKSourceConfigurationProperty | Struct | The configuration for the Amazon MSK cluster to be used as the source for a delivery stream. |
CfnDeliveryStream.OpenXJsonSerDeProperty | Struct | The OpenX SerDe. |
CfnDeliveryStream.OrcSerDeProperty | Struct | A serializer to use for converting data to the ORC format before storing it in Amazon S3. |
CfnDeliveryStream.OutputFormatConfigurationProperty | Struct | Specifies the serializer that you want Firehose to use to convert the format of your data before it writes it to Amazon S3. |
CfnDeliveryStream.ParquetSerDeProperty | Struct | A serializer to use for converting data to the Parquet format before storing it in Amazon S3. |
CfnDeliveryStream.PartitionFieldProperty | Struct | Represents a single field in a PartitionSpec . |
CfnDeliveryStream.PartitionSpecProperty | Struct | Represents how to produce partition data for a table. |
CfnDeliveryStream.ProcessingConfigurationProperty | Struct | The ProcessingConfiguration property configures data processing for an Amazon Kinesis Data Firehose delivery stream. |
CfnDeliveryStream.ProcessorParameterProperty | Struct | The ProcessorParameter property specifies a processor parameter in a data processor for an Amazon Kinesis Data Firehose delivery stream. |
CfnDeliveryStream.ProcessorProperty | Struct | The Processor property specifies a data processor for an Amazon Kinesis Data Firehose delivery stream. |
CfnDeliveryStream.RedshiftDestinationConfigurationProperty | Struct | The RedshiftDestinationConfiguration property type specifies an Amazon Redshift cluster to which Amazon Kinesis Data Firehose (Kinesis Data Firehose) delivers data. |
CfnDeliveryStream.RedshiftRetryOptionsProperty | Struct | Configures retry behavior in case Firehose is unable to deliver documents to Amazon Redshift. |
CfnDeliveryStream.RetryOptionsProperty | Struct | Describes the retry behavior in case Kinesis Data Firehose is unable to deliver data to the specified HTTP endpoint destination, or if it doesn't receive a valid acknowledgment of receipt from the specified HTTP endpoint destination. |
CfnDeliveryStream.S3DestinationConfigurationProperty | Struct | The S3DestinationConfiguration property type specifies an Amazon Simple Storage Service (Amazon S3) destination to which Amazon Kinesis Data Firehose (Kinesis Data Firehose) delivers data. |
CfnDeliveryStream.SchemaConfigurationProperty | Struct | Specifies the schema to which you want Firehose to configure your data before it writes it to Amazon S3. |
CfnDeliveryStream.SchemaEvolutionConfigurationProperty | Struct | The configuration to enable schema evolution. |
CfnDeliveryStream.SecretsManagerConfigurationProperty | Struct | The structure that defines how Firehose accesses the secret. |
CfnDeliveryStream.SerializerProperty | Struct | The serializer that you want Firehose to use to convert data to the target format before writing it to Amazon S3. |
CfnDeliveryStream.SnowflakeBufferingHintsProperty | Struct | Describes the buffering to perform before delivering data to the Snowflake destination. |
CfnDeliveryStream.SnowflakeDestinationConfigurationProperty | Struct | Configure Snowflake destination. |
CfnDeliveryStream.SnowflakeRetryOptionsProperty | Struct | Specify how long Firehose retries sending data to the New Relic HTTP endpoint. |
CfnDeliveryStream.SnowflakeRoleConfigurationProperty | Struct | Optionally configure a Snowflake role. |
CfnDeliveryStream.SnowflakeVpcConfigurationProperty | Struct | Configure a Snowflake VPC. |
CfnDeliveryStream.SplunkBufferingHintsProperty | Struct | The buffering options. |
CfnDeliveryStream.SplunkDestinationConfigurationProperty | Struct | The SplunkDestinationConfiguration property type specifies the configuration of a destination in Splunk for a Kinesis Data Firehose delivery stream. |
CfnDeliveryStream.SplunkRetryOptionsProperty | Struct | The SplunkRetryOptions property type specifies retry behavior in case Kinesis Data Firehose is unable to deliver documents to Splunk or if it doesn't receive an acknowledgment from Splunk. |
CfnDeliveryStream.TableCreationConfigurationProperty | Struct | The configuration to enable automatic table creation. |
CfnDeliveryStream.VpcConfigurationProperty | Struct | The details of the VPC of the Amazon ES destination. |
CfnDeliveryStreamProps | Struct | Properties for defining a CfnDeliveryStream. |
CloudWatchLogProcessor | Class | The data processor to extract message after decompression of CloudWatch Logs. |
CloudWatchLogProcessorOptions | Struct | Options for CloudWatchLogProcessor. |
CommonDestinationProps | Struct | Generic properties for defining a delivery stream destination. |
CommonDestinationS3Props | Struct | Common properties for defining a backup, intermediary, or final S3 destination for a Amazon Data Firehose delivery stream. |
Compression | Class | Possible compression options Amazon Data Firehose can use to compress data on delivery. |
Datadog | Class | A Datadog destination for data from a Kinesis Data Firehose delivery stream. |
DatadogEndpoint | Class | A Datadog endpoint URL for use with Kinesis Data Firehose. |
DatadogProps | Struct | Props for defining a Datadog destination of a Kinesis Data Firehose delivery stream. |
DataFormatConversionProps | Struct | Props for specifying data format conversion for Firehose. |
DataProcessorBindOptions | Struct | Options when binding a DataProcessor to a delivery stream destination. |
DataProcessorConfig | Struct | The full configuration of a data processor. |
DataProcessorIdentifier | Struct | The key-value pair that identifies the underlying processor resource. |
DataProcessorProps | Struct | Configure the LambdaFunctionProcessor. |
DecompressionProcessor | Class | The data processor to decompress CloudWatch Logs. |
DecompressionProcessorCompressionFormat | Class | Compression format for DecompressionProcessor. |
DecompressionProcessorOptions | Struct | Options for DecompressionProcessor. |
DeliveryStreamAttributes | Struct | A full specification of a delivery stream that can be used to import it fluently into the CDK application. |
DeliveryStreamGrants | Facade | Collection of grant methods for a IDeliveryStreamRef. |
DeliveryStreamProps | Struct | Properties for a new delivery stream. |
DestinationBindOptions | Struct | Options when binding a destination to a delivery stream. |
DestinationConfig | Struct | An Amazon Data Firehose delivery stream destination configuration. |
DestinationS3BackupProps | Struct | Properties for defining an S3 backup destination. |
DisableLogging | Class | Disables logging for error logs. |
DynamicPartitioningProps | Struct | Props for defining dynamic partitioning. |
EnableLogging | Class | Enables logging for error logs with an optional custom CloudWatch log group. |
HiveJsonInputFormat | Class | This class specifies properties for Hive JSON input format for record format conversion. |
HiveJsonInputFormatProps | Struct | Props for Hive JSON input format for data record format conversion. |
HttpAttribute | Struct | Describes the metadata sent to the Http endpoint destination. |
HttpBackupMode | Enum | Describes the S3 bucket backup options for the data that Kinesis Data Firehose delivers to the Http endpoint destination. |
HttpBufferingHints | Struct | The buffering options that can be used before data is delivered to the specified destination. |
HttpCompression | Enum | Kinesis Data Firehose uses the content encoding to compress the body of a request before sending the request to the destination. |
HttpEndpoint | Class | An Http destination for data from a Kinesis Data Firehose delivery stream. |
HttpEndpointConfig | Struct | Describes the configuration of the Http endpoint to which Kinesis Firehose delivers data. |
HttpEndpointProps | Struct | Props for defining an Http destination of a Kinesis Data Firehose delivery stream. |
HttpRetryOptions | Struct | Describes the retry behavior in case Kinesis Data Firehose is unable to deliver data to the specified Http endpoint destination, or if it doesn't receive a valid acknowledgment of receipt from the specified Http endpoint destination. |
IDataProcessor | Interface | A data processor that Amazon Data Firehose will call to transform records before delivering data. |
IDeliveryStream | Interface | Represents an Amazon Data Firehose delivery stream. |
IDestination | Interface | An Amazon Data Firehose delivery stream destination. |
IInputFormat | Interface | An input format to be used in Firehose record format conversion. |
ILoggingConfig | Interface | Configuration interface for logging errors when data transformation or delivery fails. |
InputFormat | Class | Represents possible input formats when performing record data conversion. |
IOutputFormat | Interface | An output format to be used in Firehose record format conversion. |
ISource | Interface | An interface for defining a source that can be used in an Amazon Data Firehose delivery stream. |
JsonParsingEngine | Class | The JSON parsing engine for MetadataExtractionProcessor. |
KinesisStreamSource | Class | An Amazon Data Firehose delivery stream source. |
LambdaFunctionProcessor | Class | Use an AWS Lambda function to transform records. |
MetadataExtractionProcessor | Class | The data processor for dynamic partitioning with inline parsing. |
MetadataExtractionProcessorOptions | Struct | Props for MetadataExtractionProcessor. |
OpenXJsonInputFormat | Class | This class specifies properties for OpenX JSON input format for record format conversion. |
OpenXJsonInputFormatProps | Struct | Props for OpenX JSON input format for data record format conversion. |
OrcCompression | Class | Possible compression options available for ORC OutputFormat. |
OrcFormatVersion | Enum | The available WriterVersions for ORC output format. |
OrcOutputFormat | Class | This class specifies properties for ORC output format for record format conversion. |
OrcOutputFormatProps | Struct | Props for ORC output format for data record format conversion. |
OutputFormat | Class | Represents possible output formats when performing record data conversion. |
ParquetCompression | Class | Possible compression options available for Parquet OutputFormat. |
ParquetOutputFormat | Class | This class specifies properties for Parquet output format for record format conversion. |
ParquetOutputFormatProps | Struct | Props for Parquet output format for data record format conversion. |
ParquetWriterVersion | Enum | The available WriterVersions for Parquet output format. |
RecordDeAggregationProcessor | Class | The data processor for multi record deaggrecation. |
RecordDeAggregationProcessorOptions | Struct | Props for RecordDeAggregationProcessor. |
S3Bucket | Class | An S3 bucket destination for data from an Amazon Data Firehose delivery stream. |
S3BucketProps | Struct | Props for defining an S3 destination of an Amazon Data Firehose delivery stream. |
SchemaConfiguration | Class | Represents a schema configuration for Firehose S3 data record format conversion. |
SchemaConfigurationBindOptions | Struct | Options when binding a SchemaConfig to a Destination. |
SchemaConfigurationFromCfnTableProps | Struct | Options for creating a Schema for record format conversion from a glue.CfnTable. |
StreamEncryption | Class | Represents server-side encryption for an Amazon Firehose Delivery Stream. |
StreamEncryptionType | Enum | Options for server-side encryption of a delivery stream. |
SubRecordType | Enum | The sub-record type to deaggregate input records. |
TimestampParser | Class | Value class that wraps a Joda Time format string. |