Para obtener capacidades similares a las de Amazon Timestream, considere Amazon Timestream LiveAnalytics para InfluxDB. Ofrece una ingesta de datos simplificada y tiempos de respuesta a las consultas en milisegundos de un solo dígito para realizar análisis en tiempo real. Obtenga más información aquí.

Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Amplíe Timestream para InfluxDB con complementos para el motor de procesamiento

El motor de procesamiento es una máquina virtual Python integrada que se ejecuta dentro de la base de datos de InfluxDB 3 en Amazon Timestream. Está disponible en las ediciones Core y Enterprise. Le permite ampliar su base de datos con código Python personalizado que puede automatizar los flujos de trabajo, transformar datos y crear puntos finales de API personalizados.

El motor de procesamiento ejecuta complementos de Python en respuesta a eventos específicos de la base de datos:

El motor incluye una caché en memoria para administrar el estado entre las ejecuciones, lo que le permite crear aplicaciones con estado directamente en la base de datos.

InfluxData complementos certificados

En el momento del lanzamiento, InfluxDB 3 incluye un conjunto de complementos prediseñados y totalmente configurables certificados por: InfluxData

Estos complementos certificados están listos para usarse y se pueden configurar mediante argumentos de activación para cumplir con sus requisitos específicos.

Tipos de complementos y especificaciones de activación

Crear desencadenadores de

Los activadores conectan los complementos a los eventos de la base de datos y definen cuándo se ejecutan. Utilice el comando influxdb3 create trigger.

Para crear un activador de escritura de datos:

# 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

Para crear un activador programado:

# 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

Para crear un activador de solicitudes 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

Acceda al punto de conexión en: https://your-cluster-endpoint:8086/api/v3/engine/webhook

Configurar activadores

Pasar argumentos a complementos

Configura el comportamiento del complemento mediante argumentos desencadenantes:

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

Los argumentos se pasan al complemento como un diccionario:

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

Comportamiento de gestión de errores

Configure cómo los desencadenadores gestionan los errores:

# 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

Ejecución asíncrona

Permita que se ejecuten varias instancias de activación de forma simultánea:

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

Administrar los activadores

Para ver los activadores de una base de datos:

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

Exclusión de tablas para los activadores de escritura

Para filtrar las tablas del código de su complemento cuando utilice 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

La implementación del complemento es la siguiente:

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

Consideraciones acerca de la implementación distribuida

En las implementaciones de varios nodos, configure los complementos según las funciones de los nodos:

Las siguientes consideraciones son importantes para las implementaciones empresariales:

Prácticas recomendadas

Supervisión de la ejecución del complemento

Consulte las tablas del sistema para supervisar el rendimiento de los complementos:

-- 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'

El motor de procesamiento proporciona una forma eficaz de ampliar la funcionalidad de InfluxDB 3 y, al mismo tiempo, mantener la lógica del procesamiento de datos cerca de los datos, lo que reduce la latencia y simplifica la arquitectura.

InfluxData complementos certificados

Amazon Timestream para InfluxDB 3 incluye un conjunto completo de complementos certificados y prediseñados que amplían la funcionalidad de la base de datos sin necesidad de desarrollo personalizado. Estos complementos son completamente configurables y están listos para usarse en el momento del lanzamiento, y ofrecen capacidades avanzadas para el procesamiento de datos, la supervisión y las alertas.

Para obtener la documentación completa y el código fuente, visita el repositorio de InfluxData complementos.

Complementos disponibles

Cronograma del complemento de migración LiveAnalytics

Migre una base de datos de Timestream para a Timestream para InfluxDB LiveAnalytics

Cómo funciona: El complemento Timestream para LiveAnalytics migración funciona al unísono con el cliente Timestream para migración. LiveAnalytics El cliente ejecuta un comando Timestream for LiveAnalytics UNLOAD para exportar una base de datos a un LiveAnalytics bucket de S3 en formato Parquet. Una vez exportados los datos, el cliente genera los archivos prefirmados URLs para Parquet e invoca el complemento de migración con los prefirmados. URLs Durante la ejecución del complemento, los objetos de S3 se recuperan del depósito de S3, se transforman en el protocolo de línea InfluxDB y se escriben en una base de datos de InfluxDB 3.

Prácticas recomendadas: El complemento Timestream for LiveAnalytics Migration debe ejecutarse en un único nodo de InfluxDB 3 Enterprise. Asegúrese de que el punto final de InfluxDB 3 utilizado con el cliente del complemento sea un punto final del nodo de proceso y no el punto final del clúster. El clúster que realiza la migración no debe realizar ingestas ni realizar consultas mientras el complemento de migración esté en ejecución, ya que esto podría provocar errores de falta de memoria.

El rendimiento de la migración depende de los recursos disponibles para el nodo InfluxDB 3 y de las características de los datos que se migran. En nuestras pruebas, observamos un rendimiento de 30 000 000 LiveAnalytics de registros migrados por hora. Su rendimiento real puede variar en función de varios factores.

Mapeo de datos: en la siguiente tabla se muestra cómo se asigna el flujo temporal de los LiveAnalytics datos a los datos del protocolo de línea.

Transformación de registros de una sola medida: el siguiente es un registro de una sola medida en Timestream para LiveAnalytics la tabla: example_table

Este registro se transformará en:

example_table,host=host1,region=us-west-2,request_id=saio3242ovnfk,measure_name=cpu_usage measure_value::double=0.66 1744908174702394001

Transformación de registros de múltiples medidas: el siguiente es un registro de múltiples medidas en Timestream para que esté LiveAnalytics en la tabla example_table con todo lo que esté a la derecha de las medidas: time

Este registro se transformará en:

example_table,host=host1,region=us-west-2,request_id=saio3242ovnfk,measure_name=metrics cpu_usage=0.66,memory_usage=0.21 1744908174702394001

Características principales:

Ejemplo de uso:

# Migrate a LiveAnalytics database to InfluxDB 3
export INFLUXDB3_HOST_URL="https://<your InfluxDB 3 URL>:<your InfluxDB 3 port>"
export INFLUXDB3_AUTH_TOKEN="<your InfluxDB 3 token>"
export INFLUXDB3_DATABASE_NAME="<your InfluxDB 3 target database>"

aws s3api create-bucket --bucket <your S3 bucket name> \
    --object-lock-enabled-for-bucket --region <your region> \
    --create-bucket-configuration LocationConstraint=<your region>

python3 liveanalytics_influxdb3_migration_client.py \
    --live-analytics-database-name <your LiveAnalytics database name> \
    --s3-bucket-name <your S3 bucket name>

Resultado: El flujo temporal de la LiveAnalytics base de datos se transforma en un protocolo de línea y se ingiere en la base de datos InfluxDB 3.

Complementos de detección de anomalías

Detección de anomalías basada en MAD

Cómo funciona: utiliza la desviación absoluta media (MAD) para establecer una línea de base sólida para el comportamiento normal. A medida que llegan nuevos datos, calcula a qué MADs distancia de la mediana se encuentra cada punto. Los puntos que superan el umbral (k * MAD) se marcan como anomalías.

Características principales:

Ejemplo de uso:

# 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

Resultado: envía notificaciones en tiempo real cuando se detectan anomalías, incluidos el nombre, el valor y la duración del campo.

Complementos de transformación de datos

Transformación básica

Cómo funciona: aplica una cadena de transformaciones a los nombres y valores de los campos. Puede procesar datos históricos en lotes (programados) o transformarlos a medida que llegan (escritura de datos). Las transformaciones se aplican en el orden especificado, lo que permite canalizaciones de datos complejas.

Características principales:

Ejemplo de uso:

# 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

Resultado: crea una nueva tabla con datos transformados, conservando las marcas de tiempo y las etiquetas originales.

Descartar muestras

Cómo funciona: agrega datos de serie temporal de alta resolución en resúmenes de menor resolución. Por ejemplo, convierte datos de 1 segundo en promedios de 1 hora. Cada punto reducido de muestreo incluye metadatos sobre el número de puntos originales comprimidos y el intervalo de tiempo cubierto.

Características principales:

Ejemplo de uso:

# 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"
  }'

Resultado: crea datos reducidos con valores agregados y columnas de metadatos que muestran el número de puntos comprimidos y el intervalo de tiempo.

Complementos de supervisión y alertas

Supervisión de los cambios de estado

Cómo funciona: realiza un seguimiento de los cambios en el valor de los campos a lo largo del tiempo y alerta cuando el número de cambios supera los umbrales configurados. Puede detectar tanto los cambios de valor (valores diferentes) como las condiciones de valores específicas (equivale a un valor objetivo). Incluye controles de estabilidad para evitar alertas por señales ruidosas.

Características principales:

Ejemplo de uso:

# 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

Resultado: las alertas incluyen el nombre del campo, el número de cambios detectados, el intervalo de tiempo y los valores de etiqueta relevantes.

Recopilador de métricas del sistema

Cómo funciona: utiliza la biblioteca psutil para recopilar métricas completas del sistema del host que ejecuta InfluxDB. Recopilación de estadísticas de CPU, memoria, disco y red en intervalos configurables. Cada tipo de métrica se puede activar o desactivar de forma independiente.

Características principales:

Ejemplo de uso:

# 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

Resultado: crea varias tablas (system_cpu, system_memory, system_disk_io, etc.) con métricas detalladas para cada subsistema.

Patrones de configuración frecuentes

Uso de archivos de configuración TOML

Para configuraciones complejas, utilice archivos TOML en lugar de argumentos en línea:

# 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

Encadenamiento de complementos

Cree canalizaciones de procesamiento de datos encadenando varios complementos:

# 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

Prácticas recomendadas para los complementos

Supervisión de la ejecución del complemento

Para supervisar el rendimiento del complemento:

-- 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';

Solución de problemas comunes

En la siguiente tabla, se muestran los problemas más comunes y las posibles soluciones.

Estos complementos certificados proporcionan una funcionalidad apta para la empresa para las necesidades habituales de procesamiento de datos de serie temporal, lo que elimina la necesidad de un desarrollo personalizado y, al mismo tiempo, mantiene la flexibilidad mediante amplias opciones de configuración. Visita el GitHubrepositorio para ver documentación detallada, ejemplos y actualizaciones.