AWS CDK Reference

Appearance

v2.274.0

MetadataExtractionProcessor Class ​

The data processor for dynamic partitioning with inline parsing.

Source
aws-kinesisfirehose/lib/processors/metadata-extraction-processor.ts
Type last changed
v2.239.0 · 19 Feb 2026

Example ​

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
// Install: dotnet add package Amazon.CDK.Lib
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
// Install: go get github.com/aws/aws-cdk-go/awscdk/v2
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,
})

Initializer ​

ts
new MetadataExtractionProcessor(options: MetadataExtractionProcessorOptions, keys: string[])
python
MetadataExtractionProcessor(options: MetadataExtractionProcessorOptions, keys: typing.List[str])
java
new MetadataExtractionProcessor(MetadataExtractionProcessorOptions options, List<String> keys)
csharp
new MetadataExtractionProcessor(MetadataExtractionProcessorOptions options, string[] keys)
go
NewMetadataExtractionProcessor(Options MetadataExtractionProcessorOptions, Keys *[]*string)

Properties ​

propsDataProcessorPropspropsDataProcessorPropspropsDataProcessorPropsPropsDataProcessorPropsPropsDataProcessorPropsrequired#
The constructor props of the DataProcessor.

Methods ​

bind(scope: Construct, options: DataProcessorBindOptions): DataProcessorConfigdef bind(self, scope: Construct, options: DataProcessorBindOptions) -> DataProcessorConfigpublic DataProcessorConfig bind(Construct scope, DataProcessorBindOptions options)public DataProcessorConfig Bind(Construct scope, DataProcessorBindOptions options)func Bind(Scope Construct, Options DataProcessorBindOptions) DataProcessorConfig#
Binds this processor to a destination of a delivery stream.

Implementers should use this method to grant processor invocation permissions to the provided stream and return the necessary configuration to register as a processor.

jq16(query: { [key: string]: string }): MetadataExtractionProcessor // chainable@staticmethod def jq16(query: typing.Mapping[str, str]) -> MetadataExtractionProcessorpublic static MetadataExtractionProcessor jq16(Map<String, String> query)public static MetadataExtractionProcessor Jq16(IDictionary<string, string> query)func Jq16(Query *map[string]*string) MetadataExtractionProcessorstatic#
Creates the inline parsing configuration with JQ 1.6 engine.
1 parameter

Parameters

  • queryqueryqueryqueryQuery — A map of partition key to jq expression.