Quix Streams

Quix Streams is a cloud-native Python library designed for processing data in Kafka. It provides a lightweight interface that combines Kafka's scalability and resiliency with a Pythonic, Pandas-like API.

Key capabilities include:

• No heavy infrastructure: Runs without a JVM, external orchestrator, or server-side engine.
• Python Ecosystem Integration: Works seamlessly with pandas, scikit-learn, TensorFlow, and PyTorch.
• Stateful Processing: Supports stateful operations using RocksDB and aggregations over tumbling and hopping time windows.
• Serialization: Supports multiple formats including JSON and Quix-specific serialization.
• Reliability: Provides "at-least-once" Kafka processing guarantees.
• Deployment: Designed for container orchestration (e.g., Kubernetes) but also supports local development and Jupyter Notebooks.

A StreamingSeries represents a single column of values within a StreamingDataFrame. They are typically generated when you access a column using sdf["column_name"] or as a result of operations performed on a StreamingDataFrame.

StreamingSeries allow you to perform column-level operations such as:

• Basic Arithmetic: add, subtract, modulo, etc.
• Comparisons: greater than, equals, is/not, etc.
• Existence Checks: checking for the presence of columns or specific values.
• Chaining: combining multiple operations together in a single pipeline.

In most cases, you will interact with StreamingSeries implicitly through StreamingDataFrame operations, similar to how one uses Pandas.

# Example of implicit StreamingSeries usage
sdf = StreamingDataFrame()
sdf["new_sum_field"] = sdf["column_c"] + sdf["column_d"] + 2

Branching occurs when a StreamingDataFrame (SDF) diverges from a single point into two or more independent processing points. This allows you to create a directed acyclic graph (DAG) of operations rather than a single linear chain.

Key Use Cases

• Multiple Topic Output: Handle different data structures or schemas by transforming the same source data in different ways before producing to different topics.
• Conditional Operations: Natively support logic where different operations are applied to different subsets of data (e.g., using .filter() to create a branch that only processes specific records).
• Consolidating Applications: Combine multiple related processing applications into a single application to reduce overhead and manage overlapping transformations in one place.

The FileSink is the base class for all community file-based sinks. It inherits from BatchingSink and implements the logic for grouping messages by topic and partition.

Each batch is serialized using the configured format before being written to the destination. The specific destination (local, Azure, S3) is determined by the subclass implementation.

Core Methods:

• setup(): An abstract method used to authenticate and validate the connection.
• write(batch: SinkBatch): Writes a batch of data using the configured format.

The InfluxDB3Source extracts data from specified InfluxDB v3 measurements and dumps them to a Kafka topic.

Processing Logic:

• It processes measurements sequentially.
• It uses a tumbling window approach based on the time_delta parameter.
• It starts querying from start_date and moves towards end_date.
• Once a measurement is complete, it moves to the next one.

Important Behavior:

• If end_date is not provided, the source will run indefinitely for a single measurement and will never process other measurements.

If you intend to perform stateful operations (such as windowed aggregations) on a StreamingDataFrame created via .concat(), you must satisfy two critical requirements:

1. Identical Partition Counts: The underlying source topics must have the same number of partitions. If they do not, the application will raise an error.
2. Consistent Partitioning Algorithm: The message keys must be distributed using the same partitioning algorithm across all source topics. If they differ, the same keys might be routed to different state stores, leading to incorrect aggregation results.

Branching requires cloning the current value at the branching node $N-1$ times (where $N$ is the number of branches). This cloning is performed using pickle, which can be a performance bottleneck or fail if data is not serializable.

To minimize performance loss:

1. Reduce value size: Use column projection to keep only necessary columns before branching. Smaller values result in lower clone costs.
2. Filter upfront: Filter values before creating branches to reduce the total volume of data that needs to be cloned.

You can detect periods of inactivity for specific Kafka message keys using the stream_timeout_ms and on_stream_timeout parameters in QuixTSDataLakeSink.

To enable this feature, you must provide both parameters. The sink uses a StreamTimeoutTracker to track the last-seen timestamp per key. When a key remains silent for the specified duration, the on_stream_timeout callback is invoked once per silence period.