Quix Streams

from quixstreams import Application

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

input_topic = app.topic("game-telemetry")
output_topic = app.topic("low-latency-players")

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

# Filter only windows where average ping is less than 4ms
sdf = sdf[sdf["ping"] < 4]

# 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("game-telemetry")
output_topic = app.topic("game-telemetry-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 a new column based on source one.
sdf["is_low_latency"] = sdf["ping"] < 4

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

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