Quix Streams

from quixstreams import Application

# Filter events with a temperature higher than a specific value
app = Application(broker_address="localhost:9092")

input_topic = app.topic("temperature-data-raw")
output_topic = app.topic("temperature-data-filtered")

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

# Use a DataFrame expression to filter events
sdf = sdf[sdf["temperature_C"] >= 100]

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

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

from quixstreams import Application

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

input_topic = app.topic("temperature-data-raw")
output_topic = app.topic("temperature-data-projected")

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

# Project sensor GPS data to a new column
sdf["gps"] = sdf.apply(
    lambda row: {"lat": row["sensor_latitude"], "long": row["sensor_longitude"]}
)

# Derive a new column based on the existing one
sdf["is_freezing"] = sdf["temperature_C"] < 0

# Extract columns from the rows before sending them to the output topic
sdf = sdf[["temperature_C", "sensor_id", "gps", "is_freezing"]]

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

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

from quixstreams import Application
from quixstreams.sinks.core.influxdb3 import InfluxDB3Sink
from quixstreams.sources import CSVSource

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

# Define a CSV source connector to read data from the file
csv_source = CSVSource(path="temperature_data_test.csv", name="temperature-csv-source")

# Pass the source to a StreamingDataFrame (SDF)
sdf = app.dataframe(source=csv_source)

# Define a sink connector for InfluxDB
influxdb_sink = InfluxDB3Sink(
    token="<token>",
    host="<host>",
    organization_id="<organization_id>",
    database="<db name>",
    measurement="temperature_F",
)

# Add a new column with temperature values converted from °C to °F
sdf["temperature_F"] = sdf["temperature_C"].apply(lambda temp: temp * 9 / 5 + 32.0)

# Send rows from the SDF back to the InfluxDB database
sdf.sink(influxdb_sink)

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