AWS Kinesis Aggregation and Deaggregation

Warnings

• gotcha The PyPI version (`pip install aws-kinesis-agg`) often lags significantly behind the latest GitHub releases. For example, PyPI currently offers 1.2.3 while GitHub has 2.0.3. Users should be aware they might install an outdated version if relying solely on `pip install`.
  Fix: Check the GitHub repository's releases for the latest version. If a newer version is needed, consider installing directly from GitHub or specifying the exact version if it eventually lands on PyPI (e.g., `pip install aws-kinesis-agg==2.0.3`).
• breaking Version 2.0.0 introduced support for AWS SDK V2. While the core API of this Python library primarily deals with data formats, this implies a shift in expected behavior or compatibility with KPL clients built with newer AWS SDKs. Users migrating from pre-2.0.0 versions should re-test their integrations.
  Fix: Review your Kinesis producer and consumer applications to ensure compatibility, especially if you are using older versions of AWS SDKs elsewhere in your stack. Test thoroughly after upgrading.
• gotcha This library is exclusively designed for records aggregated using the Kinesis Producer Library (KPL) format. It will not correctly deaggregate standard Kinesis records or records aggregated by other custom methods. Feeding non-KPL aggregated data may result in errors or malformed output.
  Fix: Before deaggregating, ensure the Kinesis record data is indeed a KPL-aggregated record. KPL records begin with a magic number (0xF3899AFC). The deaggregation functions in this library perform this check internally and will raise an `InvalidKPLRecordError` if it's not a valid KPL record.
• gotcha When consuming Kinesis records via AWS Lambda events, the `data` field for each Kinesis record in the event payload is Base64-encoded. You must decode this string back into raw bytes (`base64.b64decode()`) before passing it to `deaggregate_records` or `iter_deaggregate_records`.
  Fix: Always apply `base64.b64decode(record_data_string)` to the Kinesis record data obtained from Lambda events before attempting to deaggregate.

Install

• pip install aws-kinesis-agg Install latest PyPI version (Note: may be outdated, see warnings)

Imports

• KinesisAggregator

  from aws_kinesis_agg.aggregator import KinesisAggregator

  Class for aggregating records.
• deaggregate_records

  from aws_kinesis_agg.deaggregator import deaggregate_records

  Function to deaggregate a KPL record into a list of user records.
• iter_deaggregate_records

  from aws_kinesis_agg.deaggregator import iter_deaggregate_records

  Function to deaggregate a KPL record into an iterator of user records (memory efficient).

Quickstart

This quickstart demonstrates both aggregation and deaggregation. It shows how to use `KinesisAggregator` to combine multiple user records into a single KPL-formatted record and how to use `deaggregate_records` or `iter_deaggregate_records` to extract original user records from an aggregated KPL record. The deaggregation example includes handling base64-encoded data, common when processing Kinesis records from AWS Lambda events.

import base64
from aws_kinesis_agg.aggregator import KinesisAggregator
from aws_kinesis_agg.deaggregator import deaggregate_records, iter_deaggregate_records

# --- Aggregation Example ---
print("--- Aggregation Example ---")
aggregator = KinesisAggregator()

# Add multiple user records with partition keys and optional explicit hash keys
aggregator.add_user_record("my_app_pk", b"data_for_record_1")
aggregator.add_user_record("my_app_pk", b"data_for_record_2", explicit_hash_key="12345")
aggregator.add_user_record("another_pk", b"data_for_record_3")

# Get the aggregated record bytes, ready to be sent to Kinesis
aggregated_bytes = aggregator.get_aggregated_record_bytes()
print(f"Aggregated record size: {len(aggregated_bytes)} bytes")

# --- Deaggregation Example ---
print("\n--- Deaggregation Example ---")

# Simulate receiving an aggregated record, e.g., from an AWS Lambda Kinesis event.
# Lambda events base64-encode the data, so we simulate that.

# First, create a fresh aggregated record to ensure valid input for deaggregation
fresh_aggregator = KinesisAggregator()
fresh_aggregator.add_user_record("deagg_pk_a", b"deaggregated_data_A")
fresh_aggregator.add_user_record("deagg_pk_b", b"deaggregated_data_B", explicit_hash_key="98765")
fresh_aggregator.add_user_record("deagg_pk_c", b"deaggregated_data_C")
simulated_kpl_record_bytes = fresh_aggregator.get_aggregated_record_bytes()

# If receiving from Lambda, data would be base64 encoded:
simulated_kpl_record_base64 = base64.b64encode(simulated_kpl_record_bytes).decode('utf-8')
print(f"Simulated KPL record (base64-encoded for Lambda-like input): {simulated_kpl_record_base64[:70]}...")

# Decode the base64 data first, as Lambda would deliver it
kinesis_record_data = base64.b64decode(simulated_kpl_record_base64)

# 1. Using deaggregate_records (returns a list of UserRecord objects)
print("\nDeaggregated records (using deaggregate_records, returns a list):")
user_records_list = deaggregate_records(kinesis_record_data)
for i, record in enumerate(user_records_list):
    print(f"  Record {i+1}: PartitionKey={record.partition_key}, Data={record.data.decode()}")

# 2. Using iter_deaggregate_records (returns an iterator, memory efficient for many records)
print("\nDeaggregated records (using iter_deaggregate_records, returns an iterator):")
user_records_iterator = iter_deaggregate_records(kinesis_record_data)
for i, record in enumerate(user_records_iterator):
    print(f"  Record {i+1}: PartitionKey={record.partition_key}, Data={record.data.decode()}")