使用Quix Streams降采样数据

pip install influxdb3-python pandas quixstreams<2.5

from influxdb_client_3 import InfluxDBClient3
from quixstreams import Application
import pandas

# Instantiate an InfluxDBClient3 client configured for your unmodified database
influxdb_raw = InfluxDBClient3(
    host='cluster-host.com',
    token='DATABASE_TOKEN',
    database='
RAW_DATABASE_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.0 using influxql or sql
            table = influxdb_raw.query(
                                    query=myquery,
                                    mode='pandas',
                                    language='influxql')

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

# Send the data to a Kafka topic for the downsampling process to consumer 
def main():
    """
    Read data from the Query and publish it to Kafka
    """
    #... remaining code trunctated for brevity ...

            for index, obj in enumerate(records):
                print(obj) # Obj contains each row in the table includimng temperature
                # Generate a unique message_key for each row
                message_key = obj['machineId']
                logger.info(f'Produced message with key:{message_key}, value:{obj}')

                serialized = topic.serialize(
                    key=message_key, value=obj, headers={'uuid': str(uuid.uuid4())}
                    )

                # publish each row returned in the query to the topic 'raw-data'
                producer.produce(
                    topic=topic.name,
                    headers=serialized.headers,
                    key=serialized.key,
                    value=serialized.value,
                )

# Instantiate an InfluxDBClient3 client configured for your downsampled database.
# When writing, the org= argument is required by the client (but ignored by InfluxDB).
influxdb_downsampled = InfluxDBClient3(
    host='cluster-host.com',
    token='DATABASE_TOKEN',
    database='
DOWNSAMPLED_DATABASE_NAME
',
    org=''
)

# os.environ['localdev'] = 'true' # Uncomment if you're using local Kafka rather than Quix Cloud

# Create a Quix Streams consumer 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)

input_topic = app.topic('downsampled-data')

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

def send_data_to_influx(message):
    logger.info(f'Processing message: {message}')
    try:

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

        # Construct the points dictionary
        points = {
            'measurement': measurement_name,
            'tags': tags,
            'fields': fields,
            'time': message['time']
        }

        influxdb_downsampled.write(record=points, write_precision='ms')

sdf = app.dataframe(input_topic)
sdf = sdf.update(send_data_to_influx) # Continuously apply the 'send_data' function to each message in the incoming stream

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

git clone https://github.com/quixio/template-influxdbv3-downsampling.git