component graph aggregator

컴포넌트 — Graph Aggregator

GraphAggregator5m은 DynamoDB HISTORY#STATE snapshot을 완료된 5분 bucket 단위로 집계해 Dashboard 그래프 read model을 만드는 별도 Lambda다.

---

역할

Data Processor는 메시지별 현재 상태와 HISTORY#STATE snapshot을 만든다. GraphAggregator5m은 이 snapshot들을 읽어 Dashboard가 빠르게 그릴 수 있는 5분 집계 데이터를 생성한다.

EventBridge Scheduler (rate 5 minutes)
  -> Lambda GraphAggregator5m
      -> factory별 완료된 직전 5분 bucket 계산
      -> DynamoDB HISTORY#STATE query
      -> sensor / AI / risk / node infra avg/min/max 집계
      -> DynamoDB GRAPH#5M PutItem
      -> S3 processed_agg PutObject

---

구현 위치

apps/graph-metrics-aggregator/
├── lambda_function.py
└── aggregator/
    ├── bucket.py
    ├── dynamo.py
    ├── metrics.py
    └── s3_writer.py

Terraform 배포 정의:

infra/data-pipeline/graph_aggregator_lambda.tf

---

Bucket 기준

스케줄 실행 시 기본 동작은 현재 시각이 속한 진행 중 bucket이 아니라 닫힌 bucket을 처리하는 것이다.

예를 들어 10:10:03Z에 실행되면 10:05:00Z부터 10:09:59.999Z까지의 5분 bucket을 집계한다.

bucket_start = floor(now, 5 minutes) - 5 minutes
bucket_end   = bucket_start + 5 minutes - 1ms

이 기준 때문에 아직 진행 중인 마지막 5분 구간은 다음 스케줄까지 GRAPH#5M에 나타나지 않는다.

---

입력

GraphAggregator5m은 factory별 DynamoDB history window를 query한다.

pk = FACTORY#{factory_id}
sk between HISTORY#STATE#{bucket_start}
       and HISTORY#STATE#{bucket_end}

HISTORY#STATE는 Data Processor가 LATEST를 복사해 만든 상태 snapshot이다. factory_state, infra_state, risk, pipeline_status, latest_image_snapshot 등이 함께 들어 있을 수 있다.

---

출력

DynamoDB

pk = FACTORY#{factory_id}
sk = GRAPH#5M#{bucket_start}
item_type = GRAPH#5M
ttl = created_at + GRAPH_TTL_HOURS

GRAPH_TTL_HOURS 기본값은 48시간이다.

S3

processed_agg/{factory_id}/metrics_5m/yyyy={YYYY}/mm={MM}/dd={DD}/hh={HH}/mm={MM}.json

S3 object는 DynamoDB graph item과 같은 내용을 담되, DynamoDB key와 TTL 정책 필드는 보조 metadata로 정리한다. S3 processed_agg에는 DynamoDB TTL 필드 ttl을 저장하지 않는다.

---

집계 영역

영역	집계 필드	방식
sensor	temperature_celsius, humidity_percent, pressure_hpa	count/min/max/mean/first/last
risk	score	count/min/max/mean/first/last
ai_detection	fire_score, fall_score, bend_score	score별 통계, threshold 초과 횟수
infra	node 전체 평균 CPU/Memory/Disk	count/min/max/mean/first/last
infra.nodes[]	node_id별 CPU/Memory/Disk	node별 count/min/max/mean/first/last
quality	source count, expected count, collection rate	bucket 품질 정보

각 numeric summary는 단위, count, min/max와 발생 시각, mean, first/last와 발생 시각을 포함한다.

---

Quality와 Sample Count

quality는 그래프가 정상 수집 구간인지 판단하기 위한 메타데이터다.

Dashboard는 그래프 포인트를 단순히 "값"으로 표시하지 않고 수집 품질에 따라 신뢰도를 구분해야 한다. 예를 들어 edge-iot-publisher가 30초 동안 메시지를 보내지 못했다면 해당 5분 bucket의 source_count는 100이 아니라 90이 된다. is_partial이 true인 포인트를 그래프에서 일반 포인트와 동일하게 표시하면 관제자는 "데이터가 없어서 점수가 낮은 것"과 "실제로 이상이 발생한 것"을 구분하지 못한다. quality 객체는 이 구분을 위한 근거 데이터다.

필드	의미
source_dataset	입력 데이터셋. 현재 DynamoDB HISTORY#STATE
source_count	bucket 안에 실제 관측된 snapshot 수
expected_count	기대 sample 수. 기본 5분 / 3초 = 100
collection_rate	source_count / expected_count, 최대 1.0
missing_count	기대값 대비 누락 수
is_empty	source_count가 0인지
is_partial	source_count가 expected_count보다 작은지
infra_values_from_snapshot	infra 값이 snapshot에서 온다는 표시
source_window_start_sk	query 시작 key
source_window_end_sk	query 종료 key

GraphAggregator5m은 빈 bucket도 GRAPH#5M item으로 쓸 수 있다. 이 경우 그래프 값은 비어 있고 quality.is_empty=true, quality.is_partial=true가 된다.

---

Stale 관측 처리

Data Processor refresh가 만든 HISTORY#STATE에는 마지막 sensor/infra payload가 오래된 값일 수 있다. GraphAggregator5m은 payload의 source_timestamp가 bucket 안에 있을 때만 sensor/AI/infra 관측값으로 집계한다.

단, refresh 시점에 재계산된 risk.calculated_at이 bucket 안에 있으면 risk score는 집계한다. 그래서 데이터가 끊긴 구간에서는 센서 그래프는 비어도 risk 그래프는 data_freshness로 낮아진 점수를 보여줄 수 있다.

---

Pipeline Status와의 구분

GraphAggregator5m은 pipeline_status를 계산하지 않는다. freshness 상태는 Data Processor 내부 Pipeline Status Aggregator가 계산한다.

구분	Pipeline Status Aggregator	GraphAggregator5m
구현	data-processor 내부 모듈	별도 Lambda
입력	last_infra_state_at	HISTORY#STATE
출력	LATEST.pipeline_status, history snapshot	GRAPH#5M, processed_agg
목적	현재 데이터 단절 판단	그래프용 5분 집계

---

주요 환경 변수

변수	기본값	의미
FACTORY_IDS	factory-a,factory-b,factory-c	집계 대상
BUCKET_MINUTES	5	bucket 크기
LOOKBACK_BUCKETS	1	스케줄 실행 시 닫힌 bucket 개수
GRAPH_TTL_HOURS	48	DynamoDB GRAPH#5M TTL
EXPECTED_SAMPLE_INTERVAL_SECONDS	3	quality expected count 계산 기준
AI_SCORE_THRESHOLD	0.7	AI threshold 초과 판정
S3_OUTPUT_PREFIX	processed_agg	S3 출력 prefix

---

관련 문서

• Data Processor
• Pipeline Status Aggregator
• 데이터 생명주기