apache-airflow-providers-influxdb

import os from datetime import datetime from airflow.models.dag import DAG from airflow.providers.influxdb.operators.influxdb import InfluxDBOperator # Set environment variable for InfluxDB connection (Airflow Connection ID: influxdb_default) # Example for InfluxDB v2: influxdb://user:password@localhost:8086/?token=YOUR_TOKEN&org=YOUR_ORG&bucket=YOUR_BUCKET # If using only token (no user/pass), it might be: influxdb://localhost:8086/?token=YOUR_TOKEN&org=YOUR_ORG&bucket=YOUR_BUCKET os.environ['AIRFLOW_CONN_INFLUXDB_DEFAULT'] = os.environ.get( 'AIRFLOW_CONN_INFLUXDB_DEFAULT', 'influxdb://localhost:8086/?token=my_secret_token&org=my_org&bucket=my_bucket' # Replace with actual connection details or use Airflow UI ) with DAG( dag_id='influxdb_example_dag', start_date=datetime(2023, 1, 1), schedule_interval=None, catchup=False, tags=['influxdb', 'example'], ) as dag: write_data_to_influxdb = InfluxDBOperator( task_id='write_data', influxdb_conn_id='influxdb_default', queries=[ "_measurement='cpu_usage',host='server01' cpu=60i,usage_system=20.5", "_measurement='cpu_usage',host='server02' cpu=70i,usage_system=25.0" ], # The provider assumes these are InfluxDB line protocol strings by default. # If you need to execute Flux queries, you would pass `queries=['from(bucket:"my_bucket") |> range(start: -1h)']` # and set `query_type='flux'` if needed, though line protocol for writing is most common. ) # Example of reading data (requires a Flux query) read_data_from_influxdb = InfluxDBOperator( task_id='read_data', influxdb_conn_id='influxdb_default', queries=[ 'from(bucket:"my_bucket") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "cpu_usage")' ], query_type='flux', # results_processor=lambda results: print(f"Query results: {results}") # Uncomment to process results ) write_data_to_influxdb >> read_data_from_influxdb