WSO2 CEP 기반 실시간 데이터 콜렉터(Collector) REST API 설계 및 쿼리 샘플

/* 1. 스트림의 입출력을 정의할 데이터 메타 골격 (스키마 정의) */
@Import('METRIC:1.0.0') 
define stream input (
    meta_SERVERTIME long, 
    meta_COLLECT_TIME long, 
    meta_METRIC_NAME string, 
    meta_HOSTNAME string, 
    meta_APP_ID string, 
    correlation_METRICS double
); 
 
@Export('ALERT:1.0.0') 
define stream output (
    meta_COLLECT_TIME long, 
    meta_HOSTNAME string, 
    meta_APP_ID string, 
    meta_METRIC_NAME string, 
    correlation_METRIC double
); 
 
/* 2. 본 쿼리: 실시간 데이터 버스에서 지나가는 강물을 퍼 올려서 필터링 (Where 절 역할) */
from input#window.time(1 min)[meta_APP_ID=='collector.yarn.resourceManager.YarnClusterMetrics'] 
 
/* 3. 메모리에 체류시킨 1분 치 데이터 배열의 Max/Min 차분 집계 연산 처리 */
select 
    max(meta_COLLECT_TIME) as meta_COLLECT_TIME, 
    meta_HOSTNAME, 
    meta_APP_ID, 
    'accumulate_aggregation' as meta_METRIC_NAME, 
    (max(correlation_METRICS) - min(correlation_METRICS)) / count(*) as correlation_METRIC 
group by meta_HOSTNAME, meta_APP_ID 
having correlation_METRIC > 0 
 
/* 4. 조건에 걸러진 놈들만 붉은 사이렌을 울리기 위해 Output 배출 파이프라인으로 전송 */
insert into output;