Amazon Timestream for LiveAnalytics와 유사한 기능을 원하는 경우 Amazon Timestream for InfluxDB를 고려해 보세요. 간소화된 데이터 수집과 실시간 분석을 위한 10밀리초 미만의 쿼리 응답 시간을 제공합니다. 여기에서 자세히 알아보세요.

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

처리 엔진 플러그인으로 Timestream for InfluxDB 확장

처리 엔진은 Amazon Timestream의 InfluxDB 3 데이터베이스 내에서 실행되는 임베디드 Python 가상 머신입니다. Core edition과 Enterprise edition 모두에서 사용할 수 있습니다. 이를 통해 워크플로를 자동화하고, 데이터를 변환하고, 사용자 지정 API 엔드포인트를 생성할 수 있는 사용자 지정 Python 코드로 데이터베이스를 확장할 수 있습니다.

처리 엔진은 특정 데이터베이스 이벤트에 대한 응답으로 Python 플러그인을 실행합니다.

엔진에는 실행 간 상태를 관리하기 위한 인 메모리 캐시가 포함되어 있으므로 데이터베이스 내에서 직접 상태 저장 애플리케이션을 구축할 수 있습니다.

InfluxData 인증 플러그인

시작 시 InfluxDB 3에는 InfluxData에서 인증한 사전 구축되고 완전히 구성 가능한 플러그인 세트가 포함되어 있습니다.

이러한 인증된 플러그인은 바로 사용할 수 있으며 특정 요구 사항을 충족하도록 트리거 인수를 통해 구성할 수 있습니다.

플러그인 유형 및 트리거 사양

트리거 생성

트리거는 플러그인을 데이터베이스 이벤트에 연결하고 실행 시점을 정의합니다. influxdb3 create trigger 명령을 사용합니다.

데이터 쓰기 트리거를 생성하려면 다음을 수행하세요.

# 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

예약된 트리거를 생성하려면 다음을 수행하세요.

# 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

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

https://your-cluster-endpoint:8086/api/v3/engine/webhook에서 엔드포인트에 액세스합니다.

트리거 구성

플러그인에 인수 전달

트리거 인수를 사용하여 플러그인 동작을 구성합니다.

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

인수는 플러그인에 사전으로 전달됩니다.

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

오류 처리 동작

트리거가 오류를 처리하는 방법을 구성합니다.

# 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

비동기식 실행

여러 트리거 인스턴스가 동시에 실행되도록 허용합니다.

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

트리거 관리

데이터베이스의 트리거를 보려면 다음을 수행하세요.

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

쓰기 트리거에 대한 테이블 제외

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

플러그인 구현은 다음과 같습니다.

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

분산 배포 고려 사항

다중 노드 배포에서 노드 역할을 기반으로 플러그인을 구성합니다.

엔터프라이즈 배포에는 다음 고려 사항이 중요합니다.

모범 사례

플러그인 실행 모니터링

시스템 테이블을 쿼리하여 플러그인 성능을 모니터링합니다.

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

처리 엔진은 데이터 처리 로직을 데이터에 가깝게 유지하면서 InfluxDB 3 기능을 확장하고 지연 시간을 줄이고 아키텍처를 간소화하는 강력한 방법을 제공합니다.

InfluxData 인증 플러그인

Amazon Timestream for InfluxDB 3에는 사용자 지정 개발 없이 데이터베이스 기능을 확장하는 사전 구축되고 인증된 포괄적인 플러그인 세트가 포함되어 있습니다. 이러한 플러그인은 완전히 구성 가능하며 시작 시 사용할 준비가 되어 있어 데이터 처리, 모니터링 및 알림을 위한 고급 기능을 제공합니다.

전체 설명서와 소스 코드는 InfluxData 플러그인 리포지토리를 참조하세요.

사용 가능한 플러그인

Timestream for LiveAnalytics 마이그레이션 플러그인

LiveAnalytics용 Timestream에서 InfluxDB용 Timestream으로 데이터베이스 마이그레이션

작동 방식: Timestream for LiveAnalytics 마이그레이션 플러그인은 Timestream for LiveAnalytics 마이그레이션 클라이언트와 함께 작동합니다. 클라이언트는 Timestream for LiveAnalytics UNLOAD 명령을 실행하여 LiveAnalytics 데이터베이스를 Parquet 형식의 S3 버킷으로 내보냅니다. 데이터를 내보낸 후 클라이언트는 Parquet 파일에 대해 미리 서명된 URLs을 생성하고 미리 서명된 URLs. 플러그인 실행 중에 S3 객체는 S3 버킷에서 검색되어 InfluxDB 라인 프로토콜로 변환되고 InfluxDB 3 데이터베이스에 기록됩니다.

모범 사례: Timestream for LiveAnalytics 마이그레이션 플러그인은 단일 InfluxDB 3 Enterprise 노드에서 실행해야 합니다. 플러그인 클라이언트와 함께 사용되는 InfluxDB 3 엔드포인트가 클러스터 엔드포인트가 아닌 프로세스 노드 엔드포인트인지 확인합니다. 마이그레이션을 수행하는 클러스터는 마이그레이션 플러그인이 실행되는 동안 수집 또는 쿼리를 수행하지 않아야 합니다. 메모리 부족 오류가 발생할 수 있습니다.

마이그레이션 성능은 InfluxDB 3 노드에서 사용할 수 있는 리소스와 마이그레이션 중인 데이터의 특성에 따라 달라집니다. 테스트에서 시간당 마이그레이션된 LiveAnalytics 레코드의 처리량은 30,000,000개로 나타났습니다. 실제 성능은 여러 요인에 따라 달라질 수 있습니다.

데이터 매핑: 다음 표는 Timestream for LiveAnalytics 데이터가 라인 프로토콜 데이터에 매핑되는 방법을 보여줍니다.

단일 측정 레코드 변환: 다음은 표의 Timestream for LiveAnalytics의 단일 측정 레코드입니다example_table.

이 레코드는 다음으로 변환됩니다.

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

다중 측정값 레코드 변환: 다음은 표의 Timestream for LiveAnalytics의 다중 측정값 레코드example_table로, 측정값 오른쪽에 모든 time 것이 있습니다.

이 레코드는 다음으로 변환됩니다.

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

주요 기능:

사용 예:

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

출력: Timestream for LiveAnalytics 데이터베이스는 라인 프로토콜로 변환되고 InfluxDB 3 데이터베이스로 수집됩니다.

이상 탐지 플러그인

MAD 기반 이상 탐지

작동 방식: 중앙값 절대 편차(MAD)를 사용하여 정상 동작에 대한 강력한 기준을 설정합니다. 새 데이터가 도착하면 각 포인트의 중앙값에서 몇 개의 MAD가 떨어져 있는지 계산합니다. 임곗값(k * MAD)을 초과하는 포인트는 이상으로 플래그가 지정됩니다.

주요 기능:

사용 예:

# 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

출력: 이상이 탐지되면 필드 이름, 값 및 지속 시간을 포함한 실시간 알림을 보냅니다.

데이터 변환 플러그인

Basic Transformation

작동 방식: 필드 이름과 값에 변환 체인을 적용합니다. 기록 데이터를 배치 처리(예약)하거나 데이터가 도착하는 즉시 변환(데이터 쓰기)할 수 있습니다. 변환은 지정된 순서대로 적용되므로 복잡한 데이터 파이프라인이 허용됩니다.

주요 기능:

사용 예:

# 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

출력: 변환된 데이터를 사용하여 새 테이블을 생성하고, 원래 타임스탬프와 태그는 유지합니다.

다운샘플러

작동 방식: 고해상도 시계열 데이터를 저해상도 요약으로 집계합니다. 예를 들어, 1초 데이터를 평균 1시간으로 변환합니다. 다운샘플링된 각 포인트에는 압축된 원래 포인트 수와 해당하는 시간 범위에 대한 메타데이터가 포함됩니다.

주요 기능:

사용 예:

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

출력: 집계된 값과 압축된 포인트 수 및 시간 범위를 보여주는 메타데이터 열을 사용하여 다운샘플링된 데이터를 생성합니다.

모니터링 및 알림 플러그인

상태 변경 모니터링

작동 방식: 시간 경과에 따른 필드 값 변경을 추적하고 변경 횟수가 구성된 임곗값을 초과하면 알립니다. 값의 변화(다른 값)와 특정 값 조건(목표 값과 같음)을 모두 탐지할 수 있습니다. 노이즈 신호로 인한 알림을 방지하기 위한 안정성 검사를 포함합니다.

주요 기능:

사용 예:

# 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

출력: 알림에는 필드 이름, 탐지된 변경 사항 수, 기간 및 관련 태그 값이 포함됩니다.

시스템 지표 수집기

작동 방식: psutil 라이브러리를 사용하여 InfluxDB를 실행하는 호스트에서 포괄적인 시스템 지표를 수집합니다. 구성 가능한 간격으로 CPU, 메모리, 디스크 및 네트워크 통계를 수집합니다. 각 지표 유형은 독립적으로 활성화/비활성화할 수 있습니다.

주요 기능:

사용 예:

# 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

출력: 각 하위 시스템에 대한 자세한 지표를 포함하는 여러 테이블(system_cpu, system_memory, system_disk_io 등)을 생성합니다.

일반적인 구성 패턴

TOML 구성 파일 사용

복잡한 구성의 경우 인라인 인수 대신 TOML 파일을 사용합니다.

# 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

플러그인 연결

여러 플러그인을 연결하여 데이터 처리 파이프라인을 생성합니다.

# 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

플러그인 모범 사례

플러그인 실행 모니터링

플러그인 성능을 모니터링하려면 다음을 수행하세요.

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

일반적인 문제 해결

다음 표에는 일반적인 문제와 가능한 해결 방법이 나와 있습니다.

이러한 인증된 플러그인은 일반적인 시계열 데이터 처리 요구 사항에 맞는 엔터프라이즈 지원 기능을 제공하므로 포괄적인 구성 옵션을 통해 유연성을 유지하면서 사용자 지정 개발이 필요하지 않습니다. 자세한 설명서, 예제 및 업데이트는 GitHub 리포지토리를 참조하세요.