Python stream processing made simple

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)

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)