abstraction

Note

Maintainer notes

This module declares some important concepts and their interfaces.

• AbcRecord

• AbcBuffer

• :class:`AbcProducer

• AbcCheckpoint

• AbcConsumer

class unistream.abstraction.AbcRecord

Abstract Class for a Record to Be Sent to a Target System

In the context of this library, a “record” refers to a structured data container. This abstract class provides a foundation that you can extend to implement your own data model specific to your project. The Python community offers several excellent libraries for data modeling, including:

• dataclasses: built-in library

• attrs: mature library, trusted by NASA

• pydantic: modern library, support type hinting and validation out of the box

• sqlalchemy ORM: SQL database ORM

• django ORM: Django ORM

• pynamodb: AWS DynamoDB ORM

This abstract class should include the following attributes:

• id: unique identifier for the record.

• create_at: the ISO8601 representation of the creation time of the record.

  it has to be timezone aware.

Additionally, the class should provide the following methods:

• AbcRecord.create_at_dt(): return the timezone aware datetime object of the creation time.

• AbcRecord.serialize(): serialize the record to a string.

• AbcRecord.deserialize(): deserialize the string to a record.

property create_at_datetime: datetime

Return the datetime object of the creation time of the record.

abstract serialize() → str

Serialize the record to a string.

abstract classmethod deserialize(data: str)

Deserialize the string to a record.

class unistream.abstraction.AbcBuffer

Abstract Buffer Class for Data Producers

The abstract buffer class is designed to be used in data producer applications. Buffers play a crucial role in temporarily storing records before sending them to the target system. This allows us to optimize the utilization of network bandwidth efficiently.

Buffer Functionality

Buffers naturally follow a FIFO (First-In-First-Out) queue structure.

Fault-Tolerant Behavior

One of the key features of a buffer is its fault tolerance. It should be capable of recovering from a crash or system failure. For instance, when a record is placed in the buffer, it should be immediately persisted to ensure data durability.

Buffer Capacity

A buffer has to have these two attributes:

Parameters:

• max_records – The maximum number of records that can be stored within the buffer.

• max_size – The maximum total size, in bytes, for records that can be stored in the buffer.

When the in-memory queue reaches its maximum capacity (either in terms of records or size), the buffer will automatically write the in-memory data to persistent storage and clear the in-memory queue.

Buffer Operations

• AbcBuffer.new(): A factory method to create a new buffer instance.

• AbcBuffer.put(): Places a record into the in-memory queue.

• AbcBuffer.should_i_emit(): Checks whether the buffer should emit records.

• AbcBuffer.emit(): Emits a list of records from the buffer, following the FIFO order.

• AbcBuffer.commit(): Marks previously emitted records as no longer needed.

Who implements this

All buffer methods are for plugin/backend developers to implement. End users typically use a pre-built buffer (e.g. FileBuffer) and do not need to subclass this directly.

In summary, this abstract buffer class provides a flexible and fault-tolerant mechanism for managing data records in a data producer application.

abstract classmethod new(**kwargs)

Factory method to create a buffer. It should try to recovery unsent records from persistence layer.

abstract put(record: AbcRecord)

Put a record into the in-memory queue.

abstract should_i_emit() → bool

Identify whether the buffer should emit records.

abstract emit() → list[AbcRecord]

Emit a list of records. Older records comes first.

abstract commit()

Mark the previously emitted records as no-longer-need. Typically, it removes the records from the persistence layer.

class unistream.abstraction.AbcProducer

Abstract Class for Data Producers

A data producer is an application responsible for generating data records and seamlessly sending them to a target system. It simplifies the process for users, allowing them to focus on creating data records without needing to concern themselves with low-level details such as API calls, retry mechanisms, or fault tolerance.

A producer has to have a buffer attribute, which is an instance of a subclass of AbcBuffer.

Producer Operations

• AbcProducer.new(): A factory method to create a new producer instance.

• AbcProducer.send(): Send batch records to target system.

• AbcProducer.put(): Put the record to the buffer and smartly decide

  whether to send the records.

Who implements what

• send() and new() — Plugin/backend developers implement these to integrate with a specific streaming backend.

• put() — End users call this method to send records. It is already implemented in BaseProducer.

abstract classmethod new(**kwargs)

Factory method to create a producer.

abstract send(records: Iterable[AbcRecord])

[Plugin Developer] Send batch records to target system.

Plugin/backend developers implement this method to integrate with a specific streaming backend (e.g. Kinesis put_records, Kafka produce).

Note

You don’t need to include any logic for error handling, retry, buffer. Just think of how to send a batch of records. Those logics will be handled by the buffer and other methods.

End users do not call this method directly — call put() instead, and send will be invoked automatically when the buffer is full.

abstract put(record: AbcRecord, raise_send_error: bool = False, verbose: bool = False)

[End User API] Put the record to the buffer and smartly decide whether to send the records.

This is the main entry point for end users to send records. Already implemented in BaseProducer.

class unistream.abstraction.AbcCheckPoint

Abstract Checkpoint Class for Data Consumer

The CheckPoint class serves as a crucial component for data consumers. It stores essential information, including processing status, processing metadata, and the original record data. Its primary purpose is to ensure data integrity and achieve exactly-once processing.

CheckPoint Functionality

1. Management of Stream Pointers: Many stream systems feature a concept

   known as a “pointer” that indicates where to begin pulling data. In Kafka, this pointer is called an offset, while in Kinesis, it is referred to as a shard iterator. CheckPoint stores these pointers in the persistence layer, allowing consumers to resume from the last checkpoint in the event of a restart.

2. Batch Data Backup: When a consumer receives a batch of records,

   CheckPoint creates a short-lived backup of these records. This backup ensures that, even in scenarios where both the records and pointers are lost, the batch data can still be recovered from the checkpoint.

3. Handling Record Processing: Consumers may choose to consume batch records

   sequentially or in parallel. Before processing a record, CheckPoint sets its status as “in-progress” and locks the record to prevent other consumers from processing the same record concurrently. After processing, the checkpoint updates the record status to one of the following: “failed,” “exhausted” (retried too many times), or “succeeded,” and subsequently unlocks the record. In the event of a consumer crash during record processing, the record will be automatically unlocked after a timeout period.

CheckPoint Operations

• AbcCheckPoint.dump(): Dump the checkpoint data to the persistence layer.

• AbcCheckPoint.load(): Load the checkpoint data from the persistence layer.

• AbcCheckPoint.dump_records(): Dump the batch records data to the persistence layer.

• AbcCheckPoint.load_records(): Load the batch records data from the persistence layer. Not from the stream system.

• AbcCheckPoint.mark_as_in_progress():

• AbcCheckPoint.mark_as_failed_or_exhausted():

• AbcCheckPoint.mark_as_succeeded():

• AbcCheckPoint.dump_as_in_progress():

• AbcCheckPoint.dump_as_failed_or_exhausted():

• AbcCheckPoint.dump_as_succeeded():

Who implements what

• dump, load, dump_records, load_records, dump_as_* — Plugin/backend developers implement these to provide persistence (e.g. DynamoDB, S3, local files).

• mark_as_*, is_ready_for_next_batch, update_for_new_batch — Already implemented in BaseCheckPoint. These are framework internal methods called automatically by the consumer loop.

• get_tracker, get_not_succeeded_records — End users may call these for inspection or DLQ handling.

dump()

Dump the checkpoint data to the persistence layer.

classmethod load(**kwargs)

Load the checkpoint data from the persistence layer.

It has to handle the edge case that the checkpoint data does not exist.

dump_records(records: Iterable[AbcRecord])

Dump the batch records data to the persistence layer.

load_records(record_class: type[AbcRecord], **kwargs) → Iterable[AbcRecord]

Load the batch records data from the persistence layer. Not from the stream system.

mark_as_in_progress(record: AbcRecord, **kwargs)

Set status as in_progress and lock the record so other workers can’t process it.

Note

This method only updates the in-memory data. It is up to the developer to implement the persistence layer to persist the data.

Parameters:

record – the record we are tracking.

mark_as_failed_or_exhausted(record: AbcRecord, **kwargs)

Mark the tracker as failed or exhausted and release the lock.

Note

This method only updates the in-memory data. It is up to the developer to implement the persistence layer to persist the data.

Parameters:

record – the record we are tracking.

mark_as_succeeded(record: AbcRecord, **kwargs)

Mark the tracker as succeeded and release the lock.

Note

This method only updates the in-memory data. It is up to the developer to implement the persistence layer to persist the data.

Parameters:

record – the record we are tracking.

dump_as_in_progress(record: AbcRecord)

Dump the tracker to the persistence layer after calling BaseCheckpoint.mark_as_in_progress.

Note

It is up to the developer to implement the persistence layer to persist the data.

Parameters:

record – the record we are tracking.

dump_as_failed_or_exhausted(record: AbcRecord)

Dump the tracker to the persistence layer after calling BaseCheckpoint.mark_as_failed_or_exhausted.

Note

It is up to the developer to implement the persistence layer to persist the data.

Parameters:

record – the record we are tracking.

dump_as_succeeded(record: AbcRecord)

Dump the tracker to the persistence layer after calling BaseCheckpoint.mark_as_succeeded.

Note

It is up to the developer to implement the persistence layer to persist the data.

Parameters:

record – the record we are tracking.

class unistream.abstraction.AbcConsumer

Abstract Class for Data Consumer

A consumer is an application that continuously retrieves data from a stream system and processes it, either sequentially or in parallel. This abstract class simplifies the development work required for creating a consumer application. Users only need to focus on implementing how they want to process the data and how to handle failed data. This class automatically manages crucial aspects such as checkpointing, retries, and more.

A consumer must have a checkpoint attribute, which should be an instance of a subclass of AbcCheckpoint.

Consumer Operations

• AbcConsumer.new(): A factory method to create a new consumer instance.

• AbcConsumer.get_records(): Get records from the stream system.

• AbcConsumer.process_record(): Process a record. To indicate the processing is failed,

  it has to raise an exception.

• AbcConsumer.process_failed_record(): Process a failed record.

Who implements what

• new() and get_records() — Plugin/backend developers implement these to integrate with a specific streaming backend.

• process_record() — End users must implement this with their business logic.

• process_failed_record() — End users may override this to send failed records to a dead-letter queue (DLQ). Default is no-op.

• process_batch(), run() — End users call these methods (implemented in BaseConsumer).

abstract classmethod new(**kwargs)

Factory method to create a consumer.

abstract get_records() → Iterable[AbcRecord]

[Plugin Developer] Get records from the stream system.

Plugin/backend developers implement this method to pull records from a specific streaming backend (e.g. Kinesis get_records, Kafka poll).

abstract process_record(record: AbcRecord)

[End User] Process a record.

End users must implement this method with their business logic. To indicate the processing has failed, raise an exception.

abstract process_failed_record(record: AbcRecord)

[End User] Process a failed record.

End users may override this to send failed records to a dead-letter queue (DLQ). The default implementation is a no-op.