Use Quix Streams to downsample data

Use Quix Streams to query time series data stored in InfluxDB and written to Kafka at regular intervals, continuously downsample it, and then write the downsampled data back to InfluxDB. Quix Streams is an open source Python library for building containerized stream processing applications with Apache Kafka. It is designed to run as a service that continuously processes a stream of data while streaming the results to a Kafka topic. You can try it locally, with a local Kafka installation, or run it in Quix Cloud with a free trial.

This guide uses Python and the InfluxDB 3 Python client library, but you can use your runtime of choice and any of the available InfluxDB 3 client libraries. This guide also assumes you have already setup your Python project and virtual environment.

Pipeline architecture

The following diagram illustrates how data is passed between processes as it is downsampled:

InfluxDB v3 Source Producer

Downsampling Process

InfluxDB v3 Sink Consumer

It is usually more efficient to write raw data directly to Kafka rather than writing raw data to InfluxDB first (essentially starting the Quix Streams pipeline with the “raw-data” topic). However, this guide assumes that you already have raw data in InfluxDB that you want to downsample.

  1. Set up prerequisites

  2. Install dependencies

  3. Prepare InfluxDB buckets

  4. Create the downsampling logic

  5. Create the producer and consumer clients

    1. Create the producer client
    2. Create the consumer client

  6. Get the full downsampling code files

Set up prerequisites

The process described in this guide requires the following:

  • An InfluxDB Cloud Serverless account with data ready for downsampling.

  • A Quix Cloud account or a local Apache Kafka or Redpanda installation.

  • Familiarity with basic Python concepts.

Install dependencies

Use pip to install the following dependencies:

  • influxdb_client_3
  • quixstreams<2.5
  • pandas
pip install influxdb3-python pandas quixstreams<2.5

Prepare InfluxDB buckets

The downsampling process involves two InfluxDB buckets

. Each bucket

has a retention period that specifies how long data persists before it expires and is deleted. By using two buckets

, you can store unmodified, high-resolution data in a bucket

with a shorter retention period and then downsampled, low-resolution data in a bucket

with a longer retention period.

Ensure you have a bucket

for each of the following:

  • One to query unmodified data from
  • The other to write downsampled data to

For information about creating buckets, see Create a bucket.

Create the downsampling logic

This process reads the raw data from the input Kafka topic that stores data streamed from InfluxDB, downsamples it, and then sends it to an output topic that is used to write back to InfluxDB.

  1. Use the Quix Streams library’s Application class to initialize a connection to Apache Kafka.

    from quixstreams import Application

app = Application(consumer_group='downsampling-process', auto_offset_reset='earliest')
input_topic = app.topic('raw-data')
output_topic = app.topic('downsampled-data')

# ...

  2. Configure the Quix Streams built-in windowing function to create a tumbling window that continuously downsamples the data into 1-minute buckets.

    # ...
target_field = 'temperature' # The field that you want to downsample.

def custom_ts_extractor(value):
    # ...
    # truncated for brevity - custom code that defines the 'time_recorded'
    # field as the timestamp to use for windowing...

topic = app.topic(input_topic, timestamp_extractor=custom_ts_extractor)

sdf = (
    sdf.apply(lambda value: value[target_field])  # Extract temperature values
    .tumbling_window(timedelta(minutes=1))   # 1-minute tumbling windows
    .mean()                                  # Calculate average temperature
    .final()                                 # Emit results at window completion
)

sdf = sdf.apply(
    lambda value: {
        'time': value['end'],                  # End of the window
        'temperature_avg': value['value'],     # Average temperature
    }
)

sdf.to_topic(output_topic) # Output results to the 'downsampled-data' topic
# ...

The results are streamed to the Kafka topic, downsampled-data.

“sdf” stands for “Streaming Dataframe”.

You can find the full code for this process in the Quix GitHub repository.

Create the producer and consumer clients

Use the influxdb_client_3 and quixstreams modules to instantiate two clients that interact with InfluxDB and Apache Kafka:

  • A producer client configured to read from your InfluxDB bucket

with unmodified data and produce that data to Kafka.

  • A consumer client configured to consume data from Kafka and write the downsampled data to the corresponding InfluxDB bucket

.

Create the producer client

Provide the following credentials for the producer:

The producer queries for fresh data from InfluxDB at specific intervals. It’s configured to look for a specific measurement defined in a variable. It writes the raw data to a Kafka topic called ‘raw-data’

from influxdb_client_3 import InfluxDBClient3
from quixstreams import Application
import pandas

# Instantiate an InfluxDBClient3 client configured for your unmodified bucket
influxdb_raw = InfluxDBClient3(
    host='cloud2.influxdata.com',
    token='
API_TOKEN
',
    database='
RAW_BUCKET_NAME
'
)
# os.environ['localdev'] = 'true' # Uncomment if you're using local Kafka rather than Quix Cloud

# Create a Quix Streams producer application that connects to a local Kafka installation
app = Application(
  broker_address=os.environ.get('BROKER_ADDRESS','localhost:9092'),
  consumer_group=consumer_group_name,
  auto_create_topics=True
)

# Override the app variable if the local development env var is set to false or is not present.
# This causes Quix Streams to use an application configured for Quix Cloud
localdev = os.environ.get('localdev', 'false')

if localdev == 'false':
    # Create a Quix platform-specific application instead (broker address is in-built)
    app = Application(consumer_group=consumer_group_name, auto_create_topics=True)

topic = app.topic(name='raw-data')

## ... remaining code trunctated for brevity ...

# Query InfluxDB for the raw data and store it in a Dataframe
def get_data():
    # Run in a loop until the main thread is terminated
    while run:
        try:
            myquery = f'SELECT * FROM "{measurement_name}" WHERE time >= {interval}'
            print(f'sending query {myquery}')
            # Query InfluxDB 3 using influxql or sql
            table = influxdb_raw.query(
                                    query=myquery,
                                    mode='pandas',
                                    language='influxql')