Pour des fonctionnalités similaires à celles d'Amazon Timestream pour, pensez à Amazon Timestream LiveAnalytics pour InfluxDB. Il permet une ingestion simplifiée des données et des temps de réponse aux requêtes à un chiffre en millisecondes pour des analyses en temps réel. Pour en savoir plus, cliquez ici.

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Étendez Timestream pour InfluxDB avec des plugins du moteur de traitement

Le moteur de traitement est une machine virtuelle Python intégrée qui s'exécute dans votre base de données InfluxDB 3 dans Amazon Timestream. Il est disponible dans les éditions Core et Enterprise. Il vous permet d'étendre votre base de données avec du code Python personnalisé capable d'automatiser les flux de travail, de transformer les données et de créer des points de terminaison d'API personnalisés.

Le moteur de traitement exécute des plugins Python en réponse à des événements de base de données spécifiques :

Le moteur inclut un cache en mémoire pour gérer l'état entre les exécutions, ce qui vous permet de créer des applications dynamiques directement dans votre base de données.

InfluxData plugins certifiés

Au lancement, InfluxDB 3 inclut un ensemble de plugins prédéfinis et entièrement configurables certifiés par : InfluxData

Ces plugins certifiés sont prêts à l'emploi et peuvent être configurés via des arguments de déclenchement pour répondre à vos besoins spécifiques.

Types de plugins et spécifications des déclencheurs

Création de déclencheurs

Les déclencheurs connectent les plugins aux événements de base de données et définissent le moment de leur exécution. Utilisez la commande influxdb3 create trigger.

Pour créer un déclencheur d'écriture de données :

# Trigger on writes to a specific table
influxdb3 create trigger \
  --trigger-spec "table:sensor_data" \
  --plugin-filename "process_sensors.py" \
  --database DATABASE_NAME \
  sensor_processor

# Trigger on all table writes
influxdb3 create trigger \
  --trigger-spec "all_tables" \
  --plugin-filename "process_all_data.py" \
  --database DATABASE_NAME \
  all_data_processor

Pour créer un déclencheur planifié :

# Run every 5 minutes
influxdb3 create trigger \
  --trigger-spec "every:5m" \
  --plugin-filename "periodic_check.py" \
  --database DATABASE_NAME \
  regular_check

# Run daily at 8am (cron format with seconds)
influxdb3 create trigger \
  --trigger-spec "cron:0 0 8 * * *" \
  --plugin-filename "daily_report.py" \
  --database DATABASE_NAME \
  daily_report

Pour créer un déclencheur de requête HTTP :

# Create endpoint at /api/v3/engine/webhook
influxdb3 create trigger \
  --trigger-spec "request:webhook" \
  --plugin-filename "webhook_handler.py" \
  --database DATABASE_NAME \
  webhook_processor

Accédez au point de terminaison à l'adresse suivante : https://your-cluster-endpoint:8086/api/v3/engine/webhook

Configuration des déclencheurs

Transmission d'arguments aux plugins

Configurez le comportement du plugin à l'aide des arguments de déclenchement :

influxdb3 create trigger \
  --trigger-spec "every:1h" \
  --plugin-filename "threshold_check.py" \
  --trigger-arguments "threshold=90,notify_email=admin@example.com" \
  --database DATABASE_NAME \
  threshold_monitor

Les arguments sont transmis au plugin sous forme de dictionnaire :

def process_scheduled_call(influxdb3_local, call_time, args=None):
    if args and "threshold" in args:
        threshold = float(args["threshold"])
        email = args.get("notify_email", "default@example.com")
        # Use arguments in your logic

Comportement de gestion des erreurs

Configurez la façon dont les déclencheurs gèrent les erreurs :

# Log errors (default)
influxdb3 create trigger \
  --trigger-spec "table:metrics" \
  --plugin-filename "process.py" \
  --error-behavior log \
  --database DATABASE_NAME \
  log_processor

# Retry on error
influxdb3 create trigger \
  --trigger-spec "table:critical_data" \
  --plugin-filename "critical.py" \
  --error-behavior retry \
  --database DATABASE_NAME \
  retry_processor

# Disable trigger on error
influxdb3 create trigger \
  --trigger-spec "request:webhook" \
  --plugin-filename "webhook.py" \
  --error-behavior disable \
  --database DATABASE_NAME \
  auto_disable_processor

Exécution asynchrone

Autoriser l'exécution simultanée de plusieurs instances de déclenchement :

influxdb3 create trigger \
  --trigger-spec "table:metrics" \
  --plugin-filename "heavy_process.py" \
  --run-asynchronous \
  --database DATABASE_NAME \
  async_processor

Gérer les déclencheurs

Pour afficher les déclencheurs d'une base de données, procédez comme suit :

# Show all triggers for a database
influxdb3 show summary \
  --database DATABASE_NAME \
  --token YOUR_TOKEN

Exclusion de table pour les déclencheurs d'écriture

Pour filtrer les tableaux dans le code de votre plugin lorsque vous utilisez all_tables :

influxdb3 create trigger \
  --trigger-spec "all_tables" \
  --plugin-filename "processor.py" \
  --trigger-arguments "exclude_tables=temp_data,debug_info" \
  --database DATABASE_NAME \
  data_processor

L'implémentation du plugin est la suivante :

def process_writes(influxdb3_local, table_batches, args=None):
    excluded_tables = set(args.get('exclude_tables', '').split(','))
    
    for table_batch in table_batches:
        if table_batch["table_name"] in excluded_tables:
            continue
        # Process allowed tables

Considérations relatives au déploiement distribué

Dans les déploiements multi-nœuds, configurez les plugins en fonction des rôles des nœuds :

Les considérations suivantes sont importantes pour les déploiements en entreprise :

Bonnes pratiques

Surveiller l'exécution des plugins

Interrogez les tables du système pour surveiller les performances des plugins :

-- View processing engine logs
SELECT * FROM system.processing_engine_logs 
WHERE time > now() - INTERVAL '1 hour'
ORDER BY time DESC

-- Check trigger status
SELECT * FROM system.processing_engine_triggers
WHERE database = 'DATABASE_NAME'

Le moteur de traitement fournit un moyen puissant d'étendre les fonctionnalités d'InfluxDB 3 tout en maintenant votre logique de traitement des données proche de vos données, en réduisant la latence et en simplifiant votre architecture.

InfluxData plugins certifiés

Amazon Timestream pour InfluxDB 3 inclut un ensemble complet de plugins certifiés et prédéfinis qui étendent les fonctionnalités des bases de données sans nécessiter de développement personnalisé. Ces plugins sont entièrement configurables et prêts à être utilisés au lancement, offrant des fonctionnalités avancées pour le traitement des données, la surveillance et les alertes.

Pour obtenir une documentation complète et le code source, visitez le référentiel de InfluxData plugins.

Plugins disponibles

Plugins de détection d'anomalies

Détection d'anomalies basée sur MAD

Fonctionnement : Utilise l'écart absolu médian (MAD) pour établir une base de référence robuste pour un comportement normal. Au fur et à mesure que de nouvelles données arrivent, il calcule à quelle MADs distance de la médiane se trouve chaque point. Les points dépassant le seuil (k * MAD) sont signalés comme des anomalies.

Caractéristiques principales :

Exemple d’utilisation :

# Detect temperature anomalies in real-time
influxdb3 create trigger \
  --database sensors \
  --plugin-filename "mad_check/mad_check_plugin.py" \
  --trigger-spec "all_tables" \
  --trigger-arguments 'measurement=temperature_sensors,mad_thresholds="temp:2.5:20:5@humidity:3:30:2m",senders=slack,slack_webhook_url="YOUR_WEBHOOK"' \
  temp_anomaly_detector

# Threshold format: field:k_multiplier:window_size:trigger_condition
# temp:2.5:20:5 = temperature field, 2.5 MADs, 20-point window, alert after 5 consecutive anomalies
# humidity:3:30:2m = humidity field, 3 MADs, 30-point window, alert after 2 minutes of anomaly

Sortie : envoie des notifications en temps réel lorsque des anomalies sont détectées, notamment le nom, la valeur et la durée du champ.

Plug-ins de transformation de données

Transformation de base

Fonctionnement : applique une chaîne de transformations aux noms et aux valeurs des champs. Peut traiter les données historiques par lots (de manière planifiée) ou transformer les données au fur et à mesure de leur arrivée (écriture de données). Les transformations sont appliquées dans l'ordre indiqué, ce qui permet des pipelines de données complexes.

Caractéristiques principales :

Exemple d’utilisation :

# Transform temperature data from Celsius to Fahrenheit with field name standardization
influxdb3 create trigger \
  --database weather \
  --plugin-filename "basic_transformation/basic_transformation.py" \
  --trigger-spec "every:30m" \
  --trigger-arguments 'measurement=raw_weather,window=1h,target_measurement=weather_fahrenheit,names_transformations="Temperature Reading":"snake",values_transformations=temperature_reading:"convert_degC_to_degF"' \
  temp_converter

# Real-time field name cleaning for incoming sensor data
influxdb3 create trigger \
  --database iot \
  --plugin-filename "basic_transformation/basic_transformation.py" \
  --trigger-spec "all_tables" \
  --trigger-arguments 'measurement=raw_sensors,target_measurement=clean_sensors,names_transformations=.*:"snake alnum_underscore_only collapse_underscore"' \
  sensor_cleaner

Résultat : crée une nouvelle table avec des données transformées, en préservant les horodatages et les balises d'origine.

Sous-échantillonneur

Fonctionnement : agrège les données de séries chronologiques à haute résolution dans des résumés à faible résolution. Par exemple, convertit les données d'une seconde en moyennes d'une heure. Chaque point sous-échantillonné inclut des métadonnées concernant le nombre de points originaux compressés et la plage de temps couverte.

Caractéristiques principales :

Exemple d’utilisation :

# Downsample CPU metrics from 10-second to hourly resolution
influxdb3 create trigger \
  --database metrics \
  --plugin-filename "downsampler/downsampler.py" \
  --trigger-spec "every:1h" \
  --trigger-arguments 'source_measurement=cpu_detailed,target_measurement=cpu_hourly,interval=1h,window=6h,calculations="usage:avg.max_usage:max.total_processes:sum",specific_fields=usage.max_usage.total_processes' \
  cpu_downsampler

# HTTP endpoint for on-demand downsampling
curl -X POST http://localhost:8086/api/v3/engine/downsample \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -d '{
    "source_measurement": "sensor_data",
    "target_measurement": "sensor_daily",
    "interval": "1d",
    "calculations": [["temperature", "avg"], ["humidity", "avg"], ["pressure", "max"]],
    "backfill_start": "2024-01-01T00:00:00Z",
    "backfill_end": "2024-12-31T23:59:59Z"
  }'

Sortie : crée des données sous-échantillonnées avec des valeurs agrégées et des colonnes de métadonnées indiquant le nombre de points compressés et la plage de temps.

Plugins de surveillance et d'alerte

Moniteur de changement d'état

Fonctionnement : suit l'évolution de la valeur des champs au fil du temps et émet des alertes lorsque le nombre de modifications dépasse les seuils configurés. Peut détecter à la fois les changements de valeur (valeurs différentes) et les conditions de valeur spécifiques (égale à une valeur cible). Inclut des contrôles de stabilité pour éviter les alertes dues à des signaux bruyants.

Caractéristiques principales :

Exemple d’utilisation :

# Monitor equipment status changes
influxdb3 create trigger \
  --database factory \
  --plugin-filename "state_change/state_change_check_plugin.py" \
  --trigger-spec "every:5m" \
  --trigger-arguments 'measurement=equipment,field_change_count="status:3.temperature:10",window=15m,state_change_window=5,senders=slack,notification_text="Equipment $field changed $changes times in $window"' \
  equipment_monitor

# Real-time monitoring for specific state conditions
influxdb3 create trigger \
  --database systems \
  --plugin-filename "state_change/state_change_check_plugin.py" \
  --trigger-spec "all_tables" \
  --trigger-arguments 'measurement=service_health,field_thresholds="status:down:5@health_score:0:10",senders=pagerduty' \
  service_monitor

Résultat : les alertes incluent le nom du champ, le nombre de modifications détectées, le créneau horaire et les valeurs des balises pertinentes.

Collecteur de métriques du système

Fonctionnement : Utilise la bibliothèque psutil pour collecter des métriques système complètes auprès de l'hôte exécutant InfluxDB. Collecte les statistiques du processeur, de la mémoire, du disque et du réseau à des intervalles configurables. Chaque type de métrique peut être activé/désactivé indépendamment.

Caractéristiques principales :

Exemple d’utilisation :

# Collect all system metrics every 30 seconds
influxdb3 create trigger \
  --database monitoring \
  --plugin-filename "system_metrics/system_metrics.py" \
  --trigger-spec "every:30s" \
  --trigger-arguments 'hostname=db-server-01,include_cpu=true,include_memory=true,include_disk=true,include_network=true' \
  system_monitor

# Focus on CPU and memory for application servers
influxdb3 create trigger \
  --database app_monitoring \
  --plugin-filename "system_metrics/system_metrics.py" \
  --trigger-spec "every:1m" \
  --trigger-arguments 'hostname=app-server-01,include_cpu=true,include_memory=true,include_disk=false,include_network=false' \
  app_metrics

Sortie : crée plusieurs tables (system_cpu, system_memory, system_disk_io, etc.) avec des métriques détaillées pour chaque sous-système.

Plug-ins d'analyse prédictive

Prévisions du prophète

Fonctionnement : Utilise la bibliothèque Prophet de Facebook pour créer des modèles de prévision de séries chronologiques. Peut entraîner des modèles sur des données historiques et générer des prévisions pour les périodes futures. Les modèles tiennent compte des tendances, de la saisonnalité, des vacances et des points de changement. Supporte la persistance du modèle pour des prévisions cohérentes.

Caractéristiques principales :

Exemple d’utilisation :

# Train and forecast temperature data
influxdb3 create trigger \
  --database weather \
  --plugin-filename "prophet_forecasting/prophet_forecasting.py" \
  --trigger-spec "every:1d" \
  --trigger-arguments 'measurement=temperature,field=value,window=90d,forecast_horizont=7d,tag_values="location:seattle",target_measurement=temperature_forecast,model_mode=train,unique_suffix=seattle_v1,seasonality_mode=additive' \
  temp_forecast
# Validate temperature predictions with MAE
influxdb3 create trigger \
  --database weather \
  --plugin-filename "forecast_error_evaluator/forecast_error_evaluator.py" \
  --trigger-spec "every:1h" \
  --trigger-arguments 'forecast_measurement=temp_forecast,actual_measurement=temp_actual,forecast_field=predicted,actual_field=temperature,error_metric=mae,error_thresholds=WARN-"2.0":ERROR-"5.0",window=6h,rounding_freq=5min,senders=discord' \
  temp_forecast_check

Sortie : envoie des notifications lorsque l'erreur de prévision dépasse les seuils, y compris la valeur de la métrique d'erreur et la plage de temps affectée.

Modèles de configuration courants

Utilisation des fichiers de configuration TOML

Pour les configurations complexes, utilisez des fichiers TOML au lieu d'arguments intégrés :

# anomaly_config.toml
measurement = "server_metrics"
field = "cpu_usage"
window = "1h"
detector_type = "IsolationForestAD"
contamination = 0.1
window_size = 20
output_table = "cpu_anomalies"
senders = "slack"
slack_webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK"
notification_text = "Anomaly detected in $field: value=$value at $timestamp"

# Use TOML configuration
PLUGIN_DIR=~/.plugins influxdb3 create trigger \
  --database monitoring \
  --plugin-filename "adtk_anomaly/adtk_anomaly_detection_plugin.py" \
  --trigger-spec "every:10m" \
  --trigger-arguments "config_file_path=anomaly_config.toml" \
  cpu_anomaly_detector

Plug-ins de chaînage

Créez des pipelines de traitement des données en chaînant plusieurs plugins :

# Step 1: Transform raw data
influxdb3 create trigger \
  --database pipeline \
  --plugin-filename "basic_transformation/basic_transformation.py" \
  --trigger-spec "all_tables" \
  --trigger-arguments 'measurement=raw_sensors,target_measurement=clean_sensors,names_transformations=.*:"snake"' \
  step1_transform

# Step 2: Downsample transformed data
influxdb3 create trigger \
  --database pipeline \
  --plugin-filename "downsampler/downsampler.py" \
  --trigger-spec "every:1h" \
  --trigger-arguments 'source_measurement=clean_sensors,target_measurement=sensors_hourly,interval=1h,window=6h,calculations=avg' \
  step2_downsample

# Step 3: Detect anomalies in downsampled data
influxdb3 create trigger \
  --database pipeline \
  --plugin-filename "mad_check/mad_check_plugin.py" \
  --trigger-spec "all_tables" \
  --trigger-arguments 'measurement=sensors_hourly,mad_thresholds="value:3:20:5",senders=slack' \
  step3_anomaly

Bonnes pratiques pour les plugins

Surveiller l'exécution des plugins

Pour surveiller les performances du plugin :

-- View plugin execution logs
SELECT 
  event_time,
  trigger_name,
  log_level,
  log_text
FROM system.processing_engine_logs 
WHERE trigger_name = 'your_trigger_name'
  AND time > now() - INTERVAL '1 hour'
ORDER BY event_time DESC;

-- Monitor plugin performance
SELECT 
  trigger_name,
  COUNT(*) as executions,
  AVG(execution_time_ms) as avg_time_ms,
  MAX(execution_time_ms) as max_time_ms,
  SUM(CASE WHEN log_level = 'ERROR' THEN 1 ELSE 0 END) as error_count
FROM system.processing_engine_logs
WHERE time > now() - INTERVAL '24 hours'
GROUP BY trigger_name;

-- Check trigger status
SELECT * FROM system.processing_engine_triggers
WHERE database = 'your_database';

Résoudre les problèmes courants

Le tableau suivant présente les problèmes courants et les solutions possibles.

Ces plug-ins certifiés fournissent des fonctionnalités adaptées aux entreprises pour répondre aux besoins courants en matière de traitement de données de séries chronologiques, éliminant ainsi le besoin de développement personnalisé tout en préservant la flexibilité grâce à des options de configuration complètes. Consultez le GitHubréférentiel pour obtenir une documentation détaillée, des exemples et des mises à jour.