Para recursos semelhantes aos do Amazon Timestream para, considere o Amazon Timestream LiveAnalytics para InfluxDB. Ele oferece ingestão de dados simplificada e tempos de resposta de consulta de um dígito em milissegundos para análises em tempo real. Saiba mais aqui.

As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

Estenda o Timestream para InfluxDB com plug-ins de mecanismo de processamento

O mecanismo de processamento é uma máquina virtual Python incorporada que é executada dentro do banco de dados InfluxDB 3 no Amazon Timestream. Ele está disponível nas edições Core e Enterprise. Ele permite que você estenda seu banco de dados com código Python personalizado que pode automatizar fluxos de trabalho, transformar dados e criar endpoints de API personalizados.

O mecanismo de processamento executa plug-ins Python em resposta a eventos específicos do banco de dados:

O mecanismo inclui um cache em memória para gerenciar o estado entre as execuções, permitindo que você crie aplicativos com estado diretamente no seu banco de dados.

InfluxData plug-ins certificados

No lançamento, o InfluxDB 3 inclui um conjunto de plug-ins pré-construídos e totalmente configuráveis certificados por: InfluxData

Esses plug-ins certificados estão prontos para uso e podem ser configurados por meio de argumentos de acionadores para atender aos seus requisitos específicos.

Tipos de plug-ins e especificações de acionador

Criar acionadores

Os acionadores conectam plug-ins aos eventos do banco de dados e definem quando eles são executados. Use o comando  influxdb3 create trigger .

Para criar um acionador de gravação de dados:

# 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 criar um acionador 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 criar um acionador de solicitação 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

Acesse o endpoint em: https://your-cluster-endpoint:8086/api/v3/engine/webhook

Configurar acionadores

Passando argumentos para plug-ins

Configure o comportamento do plug-in usando argumentos de acionador:

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

Os argumentos são passados para o plugin como um dicionário:

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

Comportamento de tratamento de erros

Configure como os acionadores lidam com os erros:

# 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

Execução assíncrona

Permita que várias instâncias acionadoras sejam executadas simultaneamente:

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

Gerenciar acionadores

Para visualizar os acionadores para um banco de dados:

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

Exclusão de tabela para acionadores de gravação

Para filtrar tabelas no código do seu plug-in ao usar 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

A implementação do plug-in é a seguinte:

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

Considerações sobre a implantação distribuída

Em implantações de vários nós, configure plug-ins com base nas funções dos nós:

As considerações a seguir são importantes para implantações corporativas:

Práticas recomendadas

Monitore a execução do plugin

Consulte tabelas do sistema para monitorar o desempenho do plug-in:

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

O mecanismo de processamento fornece uma maneira poderosa de estender a funcionalidade do InfluxDB 3, mantendo sua lógica de processamento de dados próxima aos dados, reduzindo a latência e simplificando sua arquitetura.

InfluxData plug-ins certificados

O Amazon Timestream para InfluxDB 3 inclui um conjunto abrangente de plug-ins pré-criados e certificados que ampliam a funcionalidade do banco de dados sem exigir desenvolvimento personalizado. Esses plug-ins são totalmente configuráveis e prontos para uso no lançamento, fornecendo recursos avançados para processamento, monitoramento e alerta de dados.

Para obter a documentação completa e o código-fonte, visite o Repositório de InfluxData Plugins.

Plug-ins disponíveis

Plugins de detecção de anomalias

Detecção de anomalias baseada em MAD

Como funciona: usa o desvio absoluto mediano (MAD) para estabelecer uma linha de base robusta para o comportamento normal. À medida que novos dados chegam, ele calcula a que MADs distância da mediana cada ponto está. Os pontos que excedem o limite (k * MAD) são marcados como anomalias.

Principais atributos:

Exemplo 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

Saída: envia notificações em tempo real quando anomalias são detectadas, incluindo nome, valor e duração do campo.

Plug-ins de transformação de dados

Transformação básica

Como funciona: aplica uma cadeia de transformações aos nomes e valores dos campos. Pode processar dados históricos em lotes (programados) ou transformar dados à medida que chegam (gravação de dados). As transformações são aplicadas na ordem especificada, permitindo pipelines de dados complexos.

Principais atributos:

Exemplo 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

Saída: cria uma nova tabela com dados transformados, preservando os carimbos de data/hora e tags originais.

Downsampler

Como funciona: agrega dados de séries temporais de alta resolução em resumos de baixa resolução. Por exemplo, converte dados de 1 segundo em médias de 1 hora. Cada ponto com resolução reduzida inclui metadados sobre o número de pontos originais compactados e o intervalo de tempo coberto.

Principais atributos:

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

Saída: cria dados com resolução reduzida com valores agregados e colunas de metadados mostrando o número de pontos compactados e o intervalo de tempo.

Plug-ins de monitoramento e alerta

Monitor de alteração do estado

Como funciona: monitora as alterações no valor do campo ao longo do tempo e alerta quando o número de alterações excede os limites configurados. Pode detectar mudanças de valor (valores diferentes) e condições de valor específicas (igual a um valor alvo). Inclui verificações de estabilidade para evitar alertas de sinais ruidosos.

Principais atributos:

Exemplo 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

Saída: os alertas incluem o nome do campo, o número de alterações detectadas, a janela de tempo e os valores de tag relevantes.

Coletor de métricas de sistema

Como funciona: usa a biblioteca psutil para coletar métricas abrangentes do sistema do host que executa o InfluxDB. Coleta estatísticas de CPU, memória, disco e rede em intervalos configuráveis. Cada tipo de métrica pode ser ativado/desativado de forma independente.

Principais atributos:

Exemplo 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

Saída: cria várias tabelas (system_cpu, system_memory, system_disk_io etc.) com métricas detalhadas para cada subsistema.

Plug-ins de análise preditiva

Previsões do Prophet

Como funciona: usa a biblioteca Prophet do Facebook para criar modelos de previsão de séries temporais. Pode treinar modelos com base em dados históricos e gerar previsões para períodos futuros. Os modelos levam em conta tendências, sazonalidade, feriados e pontos de mudança. Oferece suporte à persistência do modelo para previsões consistentes.

Principais atributos:

Exemplo de uso:

# 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

Saída: envia notificações quando o erro de previsão excede os limites, incluindo o valor da métrica de erro e o intervalo de tempo afetado.

Padrões de configuração comuns

Utilização de arquivos de configuração TOML

Para configurações complexas, use arquivos TOML em vez de argumentos embutidos:

# 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

Encadeamento de plug-ins

Crie pipelines de processamento de dados encadeando vários plug-ins:

# 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áticas recomendadas para plug-ins

Monitore a execução do plugin

Para monitorar o desempenho do plug-in:

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

Solução de problemas comuns

A tabela a seguir mostra os problemas comuns e as possíveis soluções.

Esses plug-ins certificados fornecem funcionalidade pronta para uso corporativo para necessidades comuns de processamento de dados de séries temporais, eliminando a necessidade de desenvolvimento personalizado e mantendo a flexibilidade por meio de opções de configuração abrangentes. Visite o GitHubrepositório para obter documentação detalhada, exemplos e atualizações.