aws_kinesis_agg_record_aggregator_quickstart_with_boto3.py

1import aws_kinesis_agg.aggregator
2import boto3
3
4# Initialize the aggregator
5kinesis_aggregator = aws_kinesis_agg.aggregator.RecordAggregator()
6
7# Add user records to the aggregator
8# Arguments: Partition Key, Explicit Hash Key (optional), Data
9kinesis_aggregator.add_user_record("partition_key_1", None, b"some data 1")
10kinesis_aggregator.add_user_record("partition_key_2", None, b"some data 2")
11
12# Check if there are any aggregated records ready to be sent
13# Note: You can also check kinesis_aggregator.get_num_user_records()
14if kinesis_aggregator.get_num_user_records() > 0:
15    # Get the aggregated record
16    aggregated_record = kinesis_aggregator.clear_and_get()
17
18    # Send to Kinesis using boto3
19    kinesis_client = boto3.client('kinesis', region_name='us-east-1')
20    kinesis_client.put_record(
21        StreamName='my-kinesis-stream',
22        Data=aggregated_record.get_contents(),
23        PartitionKey=aggregated_record.get_partition_key()
24    )