Amazon Kinesis

import * as aws from "@pulumi/aws";
import * as kinesis from "@pulumi/aws/kinesis";

const stream = new aws.kinesis.Stream("data-stream", {
    name: "application-events",
    shardCount: 2,
    retentionPeriod: 24,
    streamModeDetails: {
        streamMode: "PROVISIONED",
    },
    tags: {
        Environment: "production",
    },
});

const deliveryStream = new aws.kinesis.FirehoseDeliveryStream("s3-delivery", {
    name: "event-delivery-stream",
    destination: "extended_s3",
    extendedS3Configuration: {
        roleArn: firehoseRole.arn,
        bucketArn: bucket.arn,
        prefix: "events/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/",
        errorOutputPrefix: "errors/",
        bufferingSize: 5,
        bufferingInterval: 300,
        compressionFormat: "GZIP",
    },
});

const analyticsApp = new aws.kinesisanalyticsv2.Application("analytics-app", {
    name: "stream-analytics",
    runtimeEnvironment: "FLINK-1_15",
    serviceExecutionRole: analyticsRole.arn,
    applicationConfiguration: {
        applicationCodeConfiguration: {
            codeContent: {
                s3ContentLocation: {
                    bucketArn: codeBucket.arn,
                    fileKey: "analytics-app.jar",
                },
            },
            codeContentType: "ZIPFILE",
        },
        environmentProperties: {
            propertyGroups: [{
                propertyGroupId: "ProducerConfigProperties",
                propertyMap: {
                    "aws.region": "us-west-2",
                },
            }],
        },
    },
});

const onDemandStream = new aws.kinesis.Stream("on-demand", {
    name: "on-demand-stream",
    streamModeDetails: {
        streamMode: "ON_DEMAND",
    },
    retentionPeriod: 168, // 7 days
});

const transformLambda = new aws.lambda.Function("transform", {
    runtime: "python3.11",
    handler: "index.handler",
    role: lambdaRole.arn,
    code: new pulumi.asset.AssetArchive({
        "index.py": new pulumi.asset.StringAsset(`
def handler(event, context):
    import base64
    import json
    
    output = []
    for record in event['records']:
        # Decode and transform
        payload = base64.b64decode(record['data'])
        data = json.loads(payload)
        
        # Transform logic here
        transformed = json.dumps(data)
        
        output.append({
            'recordId': record['recordId'],
            'result': 'Ok',
            'data': base64.b64encode(transformed.encode()).decode()
        })
    
    return {'records': output}
        `),
    }),
});

const deliveryStream = new aws.kinesis.FirehoseDeliveryStream("transformed", {
    name: "transformed-delivery",
    destination: "extended_s3",
    extendedS3Configuration: {
        roleArn: firehoseRole.arn,
        bucketArn: bucket.arn,
        processingConfiguration: {
            enabled: true,
            processors: [{
                type: "Lambda",
                parameters: [{
                    parameterName: "LambdaArn",
                    parameterValue: transformLambda.arn,
                }],
            }],
        },
    },
});

const consumer = new aws.kinesis.StreamConsumer("consumer", {
    name: "event-processor",
    streamArn: stream.arn,
});

const eventSourceMapping = new aws.lambda.EventSourceMapping("kinesis-trigger", {
    eventSourceArn: consumer.arn,
    functionName: processorFunction.arn,
    startingPosition: "LATEST",
    batchSize: 100,
    maximumBatchingWindowInSeconds: 10,
});