lsst-sqre/kafka-aggregator

A Kafka aggregator based on the Faust Python Stream Processing library

kafka-aggregator

A Kafka aggregator based on the Faust Python Stream Processing library.

kafka-aggregator development is based on the Safir application template.

Overview

kafka-aggregator uses Faust's windowing feature to aggregate a stream of messages from Kafka.

kafka-aggregator implements a Faust agent, a "stream processor", that adds messages from a source topic into a Faust table. The table is configured as a tumbling window with a size, representing the window duration (time interval) and an expiration time, which specifies the duration for which the data allocated to each window will be stored. Every time a window expires, a callback function is called to aggregate the messages allocated to that window. The size of the window controls the frequency of the aggregated stream.

kafka-aggregator uses faust-avro to add Avro serialization and Schema Registry support to Faust. faust-avro can parse Faust models into Avro Schemas.

See the docs for more information.

kafka-aggregator's Issues

Support for SlidingWindow

Faust appears to support a sliding window. Any chance this package currently supports a sliding window and whether there are any limitations on how large that window can get in either time-size or record size before it no longer performs?

Action Required: Fix Renovate Configuration

There is an error with this repository's Renovate configuration that needs to be fixed. As a precaution, Renovate will stop PRs until it is resolved.

Error type: undefined. Note: this is a nested preset so please contact the preset author if you are unable to fix it yourself.

a general question regarding performance

Hi,

I have a very general question regarding saving the message list in the kafka table.

I have a topic, the interval between each message is about 5 seconds. Now I want to compute mean of that topic for a window size of 4 weeks. According to your implementation, it means the table needs to be updated for every 5 seconds and there will be 4 weeks worth of data sitting in a single window.

Now my concern is about the following three lines of code:

messages = table["{{ source_topic_name }}"].value()
messages.append(message)
table["{{ source_topic_name }}"] = messages

Since messages in my case will be a huge list (483840 messages for the whole 4 weeks, and this is only for 1 user, I need to handle 1 million users, so...), my concern is that the table_change_log in Kafka will become huge as well.
My question, is Faust been optimized such that the append operation in above example ONLY append one message to the change log, or otherwise the change log actually needs to exchange 4 weeks of data points in and out for every update?

Thank you.