Python stream processing made simple

Pure Python. No JVM. No wrappers. No cross-language debugging. Use Streaming DataFrames and the whole Python ecosystem to develop stream processing pipelines in fewer lines of code.

Process data in Kafka using pure Python:

Copy

from quixstreams import Application

app = Application(broker_address="localhost:9092")

input_topic = app.topic("cardata")
output_topic = app.topic("speed-hopping-windows")

# Converts JSON to a Streaming DataFrame (SDF) tabular format
sdf = app.dataframe(input_topic)

# Calculate hopping window of 1s with 200ms steps
sdf = sdf.apply(lambda row: row["Speed"]) \
      .hopping_window(1000, 200).mean().final() 

# Send rows from SDF back to the output topic as JSON messages
sdf = sdf.to_topic(output_topic)

if __name__ == "__main__":
   app.run(sdf)

Learn more about windowing in Docs

Copy

from quixstreams import Application

app = Application(broker_address="localhost:9092")

input_topic = app.topic("cardata")
output_topic = app.topic("cardata-hardbraking")

# Converts JSON to a Streaming DataFrame (SDF) tabular format
sdf = app.dataframe(input_topic)

# Filter only windows where average brake force exceeded 50%
sdf = sdf[sdf["Brake"] > 0.5]

# Send rows from SDF back to the output topic as JSON messages
sdf = sdf.to_topic(output_topic)

if __name__ == "__main__":
   app.run(sdf)

Learn more about filtering in Docs

Copy

from quixstreams import Application

app = Application(broker_address="localhost:9092")

input_topic = app.topic("cardata")
output_topic = app.topic("cardata-with-features")

# Converts JSON to a Streaming DataFrame (SDF) tabular format
sdf = app.dataframe(input_topic)

# Project world positions columns to new column with 3 scalars
sdf["WorldPosition"] = sdf.apply(lambda row: {
    "X": row["Motion_WorldPositionX"],
    "Y": row["Motion_WorldPositionY"],
    "Z": row["Motion_WorldPositionZ"],
})

# Derive new column based on source one.
sdf["SpeedMs"] = sdf["Speed"] / 3.6

# Send rows from SDF back to the output topic as JSON messages
sdf = sdf.to_topic(output_topic)

if __name__ == "__main__":
   app.run(sdf)

Learn more about projection in Docs

Get started

Quix Streams repo

Build streaming data pipelines easily

Pure Python

No JVM. No wrappers. No cross-language debugging or complex abstraction layers. Use any package from the entire Python ecosystem.

DataFrame API

Treat data streams as continuously updating tables. Ideal for transitioning projects from Pandas or PySpark.

Flexible

No JAR files. Quix leverages Docker to simplify dependency management for stream processing pipelines.

Stateful operators

Use built-in hopping and tumbling window functions to build stateful calculations with fewer lines of code.

Scalable

Designed for efficient scaling, Quix leverages Kafka and Kubernetes to provide data partitioning, consumer groups, state management, and replication.

Fault tolerant

Guarantees reliable data delivery and robust failure recovery through data replication, service replication, changelogs and checkpointing.

Get started

Quix Streams repo

What can you build with Quix?

Check out our ready-to-run templates to kickstart your stream processing pipeline

Project template

Use case

Code snippet

Enrich data with a Redis lookup

A simple three-step pipeline that you can use as a basis for any project that requires data enrichment.

Redis

Quick Start

Quix

Project template

Use case

Code snippet

Sync data from InfluxDB V2 to InfluxDB V3

A basic two-step pipeline that you can use to keep an InfluxDB V3 database synchronized with an InfluxDB V2 bucket.

InfluxDB

Time Series Data

Quix

Project template

Use case

Code snippet

Continuously updating a vector store

A three-step pipeline that you can use to ingest embeddings into a vector database the moment new content is published.

Qdrant

SBert

LLMs

Quix

Project template

Use case

Code snippet

Hello Quix

A simple three-step pipeline that you can use as a basis for any project.

Redpanda

Quick Start

Quix

Project template

Use case

Code snippet

Real-time feature computation

A real-time feature pipeline that computes essential features that you can pass to an ML-powered trading system.

Hopsworks

Redpanda

Streamlit

Feature Engineering

Paulescu

Project template

Use case

Code snippet

AI customer support

Two sets of AI-powered chatbots are engaged in conversations with one another.

LangChain

Redpanda

InfluxDB

LLMs

Sentiment analysis

Quix

View all templates

Deploy in minutes

Deploy and manage your stream processing pipelines with Quix Cloud.

Learn more about Quix Cloud

Developer love 😻 for Quix

Open source Python client library

Support the project by starring the repo, or submit a PR to become a contributor.

Star the repo

Start building your pipeline

Create, debug and run streaming data pipelines on your local machine using the Quix CLI.

Quickstart guide

Need a helping hand? 👋

💬

Ask our community

Connect with friendly faces and get help with any problems you might be having in our Slack community.

Join us on Slack

🧑‍💻

Talk with a stream processing expert

Book a free 30 minute consultation with our experienced stream processing engineers to get all your questions answered.

Let's talk