使用 Quix Streams 对数据进行降采样

pip install influxdb-client pandas quixstreams<2.5 

from quixstreams import Application
import influxdb_client
# Create a Quix Application
app = Application(consumer_group="influxdbv2_migrate", auto_create_topics=True)
# Define the topic using the "output" environment variable
topic = app.topic(os.getenv("output", "influxv2-data"))
# Create an InfluxDB v2 client
influxdb2_client = influxdb_client.InfluxDBClient(token=os.environ["INFLUXDB_TOKEN"],
                        org=os.environ["INFLUXDB_ORG"],
                        url=os.environ["INFLUXDB_HOST"])

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

# Function to fetch data from InfluxDB
# It runs in a continuous loop, periodically fetching data based on the interval.
def get_data():
    # Run in a loop until the main thread is terminated
    while run:
        try:            
            # Query InfluxDB 2.0 using flux
            flux_query = f'''
            from(bucket: "{bucket}")
                |> range(start: -{interval})
                |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
            '''
            logger.info(f"Sending query: {flux_query}")

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

# Create a pre-configured Producer object.
with app.get_producer() as producer:
    for res in get_data():
        # Get the data from InfluxDB
        records = json.loads(res)
        for index, obj in enumerate(records):
            logger.info(f"Produced message with key:{message_key}, value:{obj}")
            # Publish the data to the Kafka topic
            producer.produce(
                topic=topic.name,
                key=message_key,
                value=obj,
            ) 

from quixstreams import Application, State
from influxdb_client import InfluxDBClient, Point

# Create a Quix platform-specific application instead
app = Application(consumer_group=consumer_group_name, auto_offset_reset="earliest", use_changelog_topics=False)

input_topic = app.topic(os.getenv("input", "input-data"))

# Initialize InfluxDB v2 client
influx2_client = InfluxDBClient(url=localhost:8086,
                                token=
API_TOKEN,
                                org=
ORG_NAME)

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

def send_data_to_influx(message: dict, state: State):
    global last_write_time_ns, points_buffer, service_start_state

    try:
        ## ... code trunctated for brevity ...

        # Check if it's time to write the batch
        # 10k records have accumulated or 15 seconds have passed
        if len(points_buffer) >= 10000 or int(time() * 1e9) - last_write_time_ns >= 15e9:
            with influx2_client.write_api() as write_api:
                logger.info(f"Writing batch of {len(points_buffer)} points written to InfluxDB.")
                write_api.write(influx_bucket, influx_org, points_buffer)

            # Clear the buffer and update the last write time
            points_buffer = []
            last_write_time_ns = int(time() * 1e9)
        
            ## ... code trunctated for brevity ...

    except Exception as e:
        logger.info(f"{str(datetime.utcnow())}: Write failed")
        logger.info(e)

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

# We use Quix Streams StreamingDataframe (SDF) to handle every message
# in the Kafka topic by writing it to InfluxDB
sdf = app.dataframe(input_topic)
sdf = sdf.update(send_data_to_influx, stateful=True)

if __name__ == "__main__":
    logger.info("Starting application")
    app.run(sdf)

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