Untuk kemampuan serupa dengan Amazon Timestream LiveAnalytics, pertimbangkan Amazon Timestream untuk InfluxDB. Ini menawarkan konsumsi data yang disederhanakan dan waktu respons kueri milidetik satu digit untuk analitik waktu nyata. Pelajari lebih lanjut di sini.

Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Perpanjang Timestream untuk InfluxDB dengan plugin mesin pemrosesan

Mesin Processing adalah mesin virtual Python tertanam yang berjalan di dalam database InfluxDB 3 Anda di Amazon Timestream. Ini tersedia dalam edisi Core dan Enterprise. Hal ini memungkinkan Anda untuk memperluas database Anda dengan kode Python kustom yang dapat mengotomatiskan alur kerja, mengubah data, dan membuat endpoint API kustom.

Mesin Processing mengeksekusi plugin Python sebagai respons terhadap peristiwa database tertentu:

Mesin menyertakan cache dalam memori untuk mengelola status di antara eksekusi, memungkinkan Anda untuk membangun aplikasi stateful langsung dalam database Anda.

InfluxData plugin bersertifikat

Saat diluncurkan, InfluxDB 3 menyertakan satu set plugin pra-bangun dan dapat dikonfigurasi sepenuhnya yang disertifikasi oleh: InfluxData

Plugin bersertifikat ini siap digunakan dan dapat dikonfigurasi melalui argumen pemicu untuk memenuhi persyaratan spesifik Anda.

Jenis plugin dan spesifikasi pemicu

Buat pemicu

Pemicu menghubungkan plugin ke peristiwa database dan menentukan kapan mereka mengeksekusi. Gunakan perintah influxdb3 create trigger.

Untuk membuat pemicu penulisan data:

# 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

Untuk membuat pemicu terjadwal:

# 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

Untuk membuat pemicu permintaan 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

Akses titik akhir di: https://your-cluster-endpoint:8086/api/v3/engine/webhook

Konfigurasikan pemicu

Melewati argumen ke plugin

Konfigurasikan perilaku plugin menggunakan argumen pemicu:

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

Argumen diteruskan ke plugin sebagai kamus:

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

Perilaku penanganan kesalahan

Konfigurasikan cara pemicu menangani kesalahan:

# 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

Eksekusi asinkron

Izinkan beberapa instance pemicu berjalan secara bersamaan:

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

Kelola pemicu

Untuk melihat pemicu database:

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

Pengecualian tabel untuk pemicu penulisan

Untuk memfilter tabel dalam kode plugin Anda saat menggunakanall_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

Implementasi plugin adalah sebagai berikut:

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

Pertimbangan penyebaran terdistribusi

Dalam penerapan multi-node, konfigurasikan plugin berdasarkan peran node:

Pertimbangan berikut penting untuk penerapan perusahaan:

Praktik terbaik

Pantau eksekusi plugin

Tabel sistem kueri untuk memantau kinerja plugin:

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

Mesin Processing menyediakan cara yang ampuh untuk memperluas fungsionalitas InfluxDB 3 sambil menjaga logika pemrosesan data Anda dekat dengan data Anda, mengurangi latensi dan menyederhanakan arsitektur Anda.

InfluxData plugin bersertifikat

Amazon TimeStream untuk InfluxDB 3 mencakup satu set lengkap plugin bersertifikat pra-bangun yang memperluas fungsionalitas basis data tanpa memerlukan pengembangan khusus. Plugin ini sepenuhnya dapat dikonfigurasi dan siap digunakan saat peluncuran, menyediakan kemampuan canggih untuk pemrosesan data, pemantauan, dan peringatan.

Untuk dokumentasi lengkap dan kode sumber, kunjungi InfluxDataPlugins Repository.

Plugin yang tersedia

Plugin deteksi anomali

Deteksi anomali berbasis MAD

Cara kerjanya: Menggunakan Median Absolute Deviation (MAD) untuk menetapkan dasar yang kuat untuk perilaku normal. Saat data baru tiba, ia menghitung berapa banyak MADs dari median setiap titik. Poin yang melebihi ambang batas (k* MAD) ditandai sebagai anomali.

Fitur utama:

Contoh penggunaan:

# 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

Keluaran: Mengirim notifikasi real-time saat anomali terdeteksi, termasuk nama bidang, nilai, dan durasi.

Plugin transformasi data

Transformasi dasar

Cara kerjanya: Menerapkan rantai transformasi ke nama dan nilai bidang. Dapat memproses data historis dalam batch (terjadwal) atau mengubah data saat tiba (data write). Transformasi diterapkan dalam urutan yang ditentukan, memungkinkan jaringan pipa data yang kompleks.

Fitur utama:

Contoh penggunaan:

# 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

Output: Membuat tabel baru dengan data yang diubah, melestarikan stempel waktu dan tag asli.

Downsampler

Cara kerjanya: Menggabungkan data deret waktu resolusi tinggi ke dalam ringkasan resolusi lebih rendah. Misalnya, mengubah data 1 detik menjadi rata-rata 1 jam. Setiap titik downsampled mencakup metadata tentang jumlah titik asli yang dikompresi dan rentang waktu yang dicakup.

Fitur utama:

Contoh penggunaan:

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

Output: Membuat data downsampled dengan nilai agregat ditambah kolom metadata yang menunjukkan jumlah titik yang dikompresi dan rentang waktu.

Memantau dan mengingatkan plugin

Monitor perubahan negara

Cara kerjanya: Melacak perubahan nilai bidang dari waktu ke waktu dan memberi peringatan ketika jumlah perubahan melebihi ambang batas yang dikonfigurasi. Dapat mendeteksi perubahan nilai (nilai berbeda) dan kondisi nilai tertentu (sama dengan nilai target). Termasuk pemeriksaan stabilitas untuk mencegah peringatan dari sinyal bising.

Fitur utama:

Contoh penggunaan:

# 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

Keluaran: Peringatan mencakup nama bidang, jumlah perubahan yang terdeteksi, jendela waktu, dan nilai tag yang relevan.

Kolektor metrik sistem

Cara kerjanya: Menggunakan pustaka psutil untuk mengumpulkan metrik sistem komprehensif dari host yang menjalankan InfluxDB. Mengumpulkan CPU, memori, disk, dan statistik jaringan pada interval yang dapat dikonfigurasi. Setiap jenis metrik dapat diaktifkan/dinonaktifkan secara independen.

Fitur utama:

Contoh penggunaan:

# 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

Output: Membuat beberapa tabel (system_cpu, system_memory, system_disk_io, dll.) Dengan metrik rinci untuk setiap subsistem.

Plugin analitik prediktif

Peramalan Nabi

Cara kerjanya: Menggunakan perpustakaan Nabi Facebook untuk membangun model peramalan deret waktu. Dapat melatih model pada data historis dan menghasilkan prediksi untuk periode waktu mendatang. Model memperhitungkan tren, musiman, liburan, dan titik perubahan. Mendukung ketekunan model untuk prediksi yang konsisten.

Fitur utama:

Contoh penggunaan:

# 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

Keluaran: Mengirim pemberitahuan saat kesalahan perkiraan melebihi ambang batas, termasuk nilai metrik kesalahan dan rentang waktu yang terpengaruh.

Pola konfigurasi umum

Menggunakan file konfigurasi TOLL

Untuk konfigurasi yang kompleks, gunakan file TOMM alih-alih argumen sebaris:

# 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

Plugin rantai

Buat pipeline pemrosesan data dengan merantai beberapa plugin:

# 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

Praktik terbaik untuk plugin

Pantau eksekusi plugin

Untuk memantau kinerja 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';

Memecahkan masalah umum

Tabel berikut menunjukkan masalah umum dan solusi yang mungkin.

Plugin bersertifikat ini menyediakan fungsionalitas siap perusahaan untuk kebutuhan pemrosesan data deret waktu umum, menghilangkan kebutuhan untuk pengembangan khusus sambil mempertahankan fleksibilitas melalui opsi konfigurasi yang komprehensif. Kunjungi GitHubrepositori untuk dokumentasi terperinci, contoh, dan pembaruan.