DARAM BLOG

Anomaly detection 종류

1. Supervised Anomaly Detection  : 트레이닝 데이터에 비정상,정상 Label이 모두 적절히존재했을 때
2. Semi-supervised Anomaly Detection : 정상 데이터만을 가지고 학습한 경우 (비정상 라벨이 심각하게 적은경우)
3. Unsupervised Anomaly Detection : 트레이닝 데이터에 정상, 비정상 Labeling이 안 된 경우

X_train = x_m
y_train = y_m  # 라벨
X_test = x_f
y_test = y_f # 라벨
X_val = x_w
Y_val = y_w # 라벨

gmm_clf = GaussianMixture(covariance_type='full', n_components=3 ,max_iter=int(1e5))  # Obtained via grid search
# covariance_type : full  tied  diag  spherical
gmm_clf.fit(X_train)
log_probs_val = gmm_clf.score_samples(X_val)
isotonic_regressor = IsotonicRegression(out_of_bounds='clip')
isotonic_regressor.fit(log_probs_val, Y_val)  

log_probs_test = gmm_clf.score_samples(X_test)
test_probabilities = isotonic_regressor.predict(log_probs_test)
test_predictions = [1 if prob >= 0.5 else 0 for prob in test_probabilities]

cm = confusion_matrix(y_test,test_predictions )
print(cm)
print(accuracy_score(y_test,test_predictions))
sns.heatmap(cm,annot=True,fmt='g')

covariance_type 매개변수에 다음 값 중 하나를 설정  

1. "spherical”: 모든 클러스터가 원형이지만 지름은 다를 수 있음(분산이 다름) 
2. "diag”: 클러스터는 크기에 상관없이 어떤 타원형도 가능하지만 타원의 축은 좌표 축과 나란핬야 함(공분산 행렧이 대각 행렧이어야 함)  
3. "tied”: 모듞 클러스터가 동일한 타원 모양, 크기, 방향을 가짐(모든 클러스터는 동일핚 공분산 행렧을 공유)  
4. “full” : covariance_type 매개변수의 기본값은 "full＂ 인데 각 클러스터는 모양, 크기, 방향에 제약이 없음(각자 제약이 없는 공분산 행렧을 가짐)

1. isolation forest 로 정상에 가까운 클러스터, 비정상에 가까운 클러스터 이렇게 두개 묶음 만들기

2.  데이터 train_정상/train_비정상/test_정상/test_비정상 이케 데이터 나눠준다

3. train_정상 만 가지고 lstm autoencoder 학습시키고 테스트 데이터 넣어서 threshold 를 구해준다 

=> 이 스레숄드가 뭐나면? 재구성손실값임. 한마디로 정상치만 가지고 학습시켰을때 똑같이 정상치라면 재구성할떄 오류가 많이 나지않음.스레숄드가 낮게나온다는 뜻임. 그렇다면 정상, 그런데 완전 다른 그림을 학습시킨다면 재구성할때 오차가 엄청날것임. 스레숄드가 엄청 크게 나온다는뜻. 스레숄드보다 높은값이 나온것은 정상이랑 많이 다른양상이라는 뜻이므로 anomaly 로 구분한다.

4. 이렇게 전체 데이터를 집어넣어봐서 스레숄드보다 높은값을 보이면 (재구성손실이 많다면) anomaly 로 라벨링하고, 스레숄드보다 낮은값을 보이면(재구성손실이 적다면, 정상치 트레인데이터와 비슷한양상을 보인다면) 정상값으로 라벨링한다.

스레숄드 구하는법은 아래와같음

** 정밀도 ( precision) : 참이라고 예측한 데이터중 실제로 참인 데이터 

**제현율 ( RECALL ) : 실제로 참인 데이터중에서 참이라고 예측한 데이터

긁어옴)) : 재현율이 높지만 정밀도가 낮다는 것은 찾은 데이터의 수가 많지만 그 중 실제 찾으려는 대상의 비율은 낮았다는 것을 의미합니다. 이와 반대로 정밀도가 높지만 재현율이 낮다는 것은 찾은 데이터의 수는 작지만 그 중 실제 찾으려는 대상의 비율은 높다는 것을 의미합니다. 이상적으로는, 정밀도와 재현율이 모두 높은 것이 좋습니다. 이는 찾은 데이터 수가 많으며 그 중 실제 찾으려는 대상의 비율이 높다는 것을 의미합니다. 그렇기 때문에 threshold 값은 이 정밀도와 재현률이 둘다 좋은지점을 찾아 threshold 라고 정의해줍니다.

참조 : 

https://www.koreascience.or.kr/article/CFKO202125036427364.pdf (국내학회)

https://towardsdatascience.com/lstm-autoencoder-for-extreme-rare-event-classification-in-keras-ce209a224cfb

https://velog.io/@jaehyeong/LSTM-Autoencoder-for-Anomaly-Detection 

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import math
from pylab import rcParams
from collections import Counter
import tensorflow as tf
from tensorflow.keras import Model ,models, layers, optimizers, regularizers
from tensorflow.keras.callbacks import ModelCheckpoint
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from sklearn import metrics
%matplotlib inline
from sklearn.ensemble import IsolationForest
from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler
from mpl_toolkits.mplot3d import Axes3D
import pickle
import glob
from tensorflow.python.keras.metrics import Metric
import sklearn

# lsmt ae 에서 사용할 함수 
# LSTM 모델은 (samples, timesteps, feature)에 해당하는 3d 차원의 shape을 가지므로, 데이터를 시퀀스 형태로 변환한다.
def temporalize(X, y, timesteps): 
	output_X = []  
	output_y = []
	for i in range(len(X) - timesteps - 1):  
		t = []
		for j in range(1, timesteps + 1):
			# Gather the past records upto the lookback period
			t.append(X[[(i + j + 1)], :])
		output_X.append(t)
		output_y.append(y[i + timesteps + 1])
	return np.squeeze(np.array(output_X)), np.array(output_y)

content={}
#데이터별로 이상치 index를 담아줄 딕셔너리. 마지막에 anomaly라고 라벨링된 index만 뽑아서 "데이터이름" : [1,2,4...] 이런식으로 key,value 값을 저장해줄것.
parameter = [0.01 , 0.015, 0.02, 0.025, 0.03]

for k in filenames:
    for j in parameter:
        print(j)
        df = pd.read_csv(str(k))
        
        
        # 1차 라벨링 부분
        #isolationForest 적용 + 그림으로 보이기 ---------------------------------------------------------
        
        from sklearn.ensemble import IsolationForest
        clf=IsolationForest(n_estimators=50, max_samples=50, contamination = float(j), 
                                max_features=1.0, bootstrap=False, n_jobs=-1, random_state=None, verbose=0)
        clf.fit(df)
        pred = clf.predict(df)
        df['anomaly'] = pred  # isolation forest로 1차 라벨링된것을 새피처로써 추가함.
        outliers = df.loc[df['anomaly'] == -1]
        outlier_index = list(outliers.index) # 1차 라벨링해서 이상치값의 index 만 outlier_index 리스트로 만듬. 
        pca = PCA(n_components = 5) # isolation 적용한후, 피처개수가 너무많아서 차원축소해준다. 

        scaler = StandardScaler()
        X = scaler.fit_transform(df)
        X_reduce = pca.fit_transform(X)
        fig = plt.figure()
        ax = fig.add_subplot(111)
        ax.scatter(X_reduce[:, 0], X_reduce[:, 1], s=4, lw=1, label="inliers",c="green")
        ax.scatter(X_reduce[outlier_index,0],X_reduce[outlier_index,1],
                   lw=1, s=4, c="red", label="outliers")
        ax.legend()
        plt.show()
        print("isolation forest 적용후값  : " + str(Counter(pred)))
        
         
        # 2차 라벨링 부분 
        # Train/valid data split -------------------------------------------------------------------------------
        
        input_x = X_reduce   # isolation forest 로 1차 라벨링한값을 차원축소한후 lstm ae 의 인풋데이터로 사용할것
        input_y = []
        # 1차 라벨링에서 1 정상 -1 비정상으로 1차 라벨링된값을 => 1 비정상 0 정상으로 숫자만 바꿔줌
        for i in range(len(input_x)):   # 1 , -1
            if i in outlier_index:
                input_y.append(1) # 비정상
            else :
                input_y.append(0) # 정상
        input_y = np.array(input_y)  
        
        timesteps = 3
        x, y = temporalize(input_x, input_y, timesteps)   # LSTM 모델은 (samples, timesteps, feature)에 해당하는 3d 차원의 shape을 가지므로, 데이터를 시퀀스 형태로 변환한다.


        # Split into train and valid
        x_train, x_valid, y_train, y_valid = train_test_split(x, y, test_size=0.2)
        n_features = input_x.shape[1]
        
        # For training the autoencoder, split 0 / 1
        x_train_y0 = x_train[y_train == 0] # 정상데이터 y0
        x_train_y1 = x_train[y_train == 1] # 비정상데이터 y1
        x_valid_y0 = x_valid[y_valid == 0]
        x_valid_y1 = x_valid[y_valid == 1]

        x_y0 = x[y==0] # 정상데이터 y0
        x_y1 = x[y==1] # 비정상데이터 y1
        
        def flatten(X):
            flattened_X = np.empty((X.shape[0], X.shape[2]))  # sample x features array.
            for i in range(X.shape[0]):
                flattened_X[i] = X[i, (X.shape[1]-1), :]
            return(flattened_X)
        def scale(X, scaler):
            for i in range(X.shape[0]):
                X[i, :, :] = scaler.transform(X[i, :, :])  
            return X

        scaler = StandardScaler().fit(flatten(x_train_y0))
        x_train_y0_scaled = scale(x_train_y0, scaler)
        x_valid_scaled = scale(x_valid, scaler)
        x_valid_y0_scaled = scale(x_valid_y0, scaler)
        x_total_scaled = scale(x, scaler)   
        x_y0_sacled = scale(x_y0, scaler)

        # LSTM Autoencoder 적용 -------------------------------------------------------------------------------
        epochs = 5
        batch = 128
        lr = 0.001
        lstm_ae = models.Sequential()
        # Encoder
        lstm_ae.add(layers.LSTM(32, activation='relu', input_shape=(timesteps, n_features), return_sequences=True))
        lstm_ae.add(layers.LSTM(16, activation='relu', return_sequences=False))
        lstm_ae.add(layers.RepeatVector(timesteps))
        # Decoder
        lstm_ae.add(layers.LSTM(16, activation='relu', return_sequences=True))
        lstm_ae.add(layers.LSTM(32, activation='relu', return_sequences=True))
        lstm_ae.add(layers.TimeDistributed(layers.Dense(n_features)))
        # compile
        lstm_ae.compile(loss='mse', optimizer=optimizers.Adam(lr))
        # fit
        history = lstm_ae.fit(x_y0_sacled, x_y0_sacled,
                             epochs=epochs, batch_size=batch,
                             validation_data=(x_valid_y0_scaled, x_valid_y0_scaled))
        
        
        
        # normal/anomaly 를 구분해주는 기준 threshold 를 구함  -------------------------------------------------------

        valid_x_predictions = lstm_ae.predict(x_valid_scaled)
        mse = np.mean(np.power(flatten(x_valid_scaled) - flatten(valid_x_predictions), 2), axis=1)

        error_df = pd.DataFrame({'Reconstruction_error':mse, 
                                 'True_class':list(y_valid)})
        precision_rt, recall_rt, threshold_rt = sklearn.metrics.precision_recall_curve(error_df['True_class'], error_df['Reconstruction_error'])
        
        
        # precision/recall 값
        index_cnt = [cnt for cnt, (p, r) in enumerate(zip(precision_rt, recall_rt)) if p==r][0]
        #print('precision: ',precision_rt[index_cnt],', recall: ',recall_rt[index_cnt])

        # threshold 값
        threshold_fixed = threshold_rt[index_cnt]
        #print('threshold: ',threshold_fixed)


        # threshold 보다 낮으면 정상(normal), 높으면 이상(anomaly)으로 판단한다.
        # 정밀도 ( precision) : 참이라고 예측한 데이터중 실제로 참인 데이터 (참 / 참인데참,거짓인데참)
        # 제햔율 ( RECALL ) : 실제로 참인 데이터중에서 참이라고 예측한 데이터? (참/ 참인데참, 참인데 거짓))

        
        
        # threshold 값을 기준으로 normal/anomlay 인지 라벨링해줌 ----------------------------------------------------------------
        test_x_predictions = lstm_ae.predict(x_total_scaled) 
        mse = np.mean(np.power(flatten(x_total_scaled) - flatten(test_x_predictions), 2), axis=1)
        error_df = pd.DataFrame({'Reconstruction_error': mse,
                                 'True_class': y.tolist()})
        groups = error_df.groupby('True_class')
        fig, ax = plt.subplots()

        for name, group in groups:
            ax.plot(group.index, group.Reconstruction_error, marker='o', ms=3.5, linestyle='',
                    label= "Break" if name == 1 else "Normal")
        ax.hlines(threshold_fixed, ax.get_xlim()[0], ax.get_xlim()[1], colors="r", zorder=100, label='Threshold')
        ax.legend()
        plt.title("Reconstruction error for different classes")
        plt.ylabel("Reconstruction error")
        plt.xlabel("Data point index")
        plt.show();
        
        
        # classification by threshold
        pred_y = [1 if e > threshold_fixed else 0 for e in error_df['Reconstruction_error'].values]
        LABELS = ['Normal', 'Break']
        conf_matrix = sklearn.metrics.confusion_matrix(error_df['True_class'], pred_y)
        plt.figure(figsize=(7, 7))
        sns.heatmap(conf_matrix, xticklabels=LABELS, yticklabels=LABELS, annot=True, fmt='d')
        plt.title('Confusion Matrix')
        plt.xlabel('Predicted Class'); plt.ylabel('True Class')
        plt.show()


        # TN/FP/FN 구하기 ------------------------------------------------------------------------------------
        # 여기서 true 값은 1차 라벨링값 predict 는 2차 라벨링값
        
        df_p = pd.DataFrame(pred_y, columns=['anomaly'])
        df_t = pd.DataFrame(error_df['True_class'].tolist(), columns=['anomaly'])
        
        anomaly_list_p = df_p[df_p['anomaly']==1].index
        anomaly_list_T = df_t[df_t['anomaly']==1].index
        anomaly_list_p = (anomaly_list_p +2).to_list()
        anomaly_list_T = (anomaly_list_T +2).to_list()

        TN = list(set(anomaly_list_T) & set(anomaly_list_p)) 
        FP = list(set(anomaly_list_p) - set(anomaly_list_T))
        FN = list(set(anomaly_list_T) - set(anomaly_list_p))
        
                          
        # anomaly index 만 리스트로 만들어서 dictionary 에 value 로 저장. ------------------------------------------------------------------------------------------
        isolationforest_parameter = j
        str_isolationforest_parameter = str(isolationforest_parameter)[2:]
        if(len(str_isolationforest_parameter) != 5  ):
            for i in range(5 - len(str_isolationforest_parameter)):
                str_isolationforest_parameter=str_isolationforest_parameter + '0'
            
        
        
        Save_dict_name = str(k)[7:20]
        content[Save_dict_name +'_p_'+ str(str_isolationforest_parameter)] = TN+FP+FN
        
        
        with open('anomaly_index.pickle','wb') as fw:
            pickle.dump(content, fw)
            
        file = open("anomaly_index.pickle",'rb')
        content = pickle.load(file)
        content.keys()

elastalert-test-rule example_rules/d.yaml

#이건 테스트 알람이가진않는다.

elastalert --verbose --rule example_rules/d.yaml

#이건 실행

**위치는 config.yaml이들어있는 파일에서 터미널열고실행해야함.

(config.yaml이랑 example_rules같이있는 파일)

테스트실행으로는 됬다 안됬다만 나타내고 알람이 안와서 제대로 됬는지 알수가없어서 알람을 받기위해 사용한 방법

kibana ---> elastalert_status_status index에 들어가보면 발생한 로그들이 저장되어있었다. 

(키바나 인덱스 들어가보면 로그인덱스찾아보기)

터미널을 여러개열고 ~~.yaml을 띄워둔다. 

아래는 config.yaml

rules_folder: example_rules
run_every:
  minutes: 1
buffer_time:
  minutes: 15
es_host: 203.246.~~  #이부분은 맞게고치기
es_port: 39200    #이부분은 맞게고치기
writeback_index: elastalert_status
alert_time_limit:
  days: 2

example_rules안에서 여러개 띄워준 터미널들 예시 (아래링크에서 rule들 참조하기)

https://elastalert.readthedocs.io/en/latest/running_elastalert.html#running-elastalert

d.yaml

name: Example frequency rule
type: flatline
index: logstash-*
timeframe:
- hours: 4
use_count_query: true
query_delay:
- minutes: 5
alert:
- "email"
- "slack"
email:
- "bbbbbbbbb@naver.com"
slack:
slack_webhook_url: "https://hooks.slack.com/services/T~~~~~~~~~~~~~~~~~U~~~9B1OvKw"
slack_username_override: 'h~~~~~~~~~~~~'
slack_channel_override: '#elastalert'
slack_emoji_override: ':pizza:'

ㄴㄴslack은 webhook연결참조하기~!!  참조로 이메일은 알람이 안뜨고 슬렉에서만왔다.

이렇게  rule들을 많이 띄워준다. 

그리고 마지막으로 이 yaml은 발생건수가 100000000개 이하이면 알람이온다!!!!!!!!!!!!

use_terms_query를 사용해서 필터를 안써도된다. 

이렇게 해주면 좀 기다리다보면 슬렉에서 알람이온다. 

slack.yaml

name: flaterrornumber
type: flatline
index: logstash-*
threshold: 10000000000
timeframe:
  hours: 1
use_count_query: true
use_terms_query: false
doc_type: "doc"

alert:
- "email"
- "slack"

email:
- "bbbbbbbbbbbb@naver.com"

slack:
slack_webhook_url: "https://hooks.slack.com/services/T0136~~~~~~~~~~~~~~~~2"
slack_username_override: 'hyo~~~~~~~~~~57'
slack_channel_override: '#elastalert'
slack_emoji_override: ':pizza:'

<파일째 원격으로 다운로드받기>

•  -r 은 파일안에 속해있는것들 죄다 복사하라는 옵션이고
•  -p는 32222포트를 의미 이옵션을 주지않으면 기본포트 22로 작동
•  capstone7은 아이디
• 204.243.112.129:~/ELK/data   이건 주소 : 원격파일위치 

scp -r -P 32222 capstone7@204.243.112.129:~/ELK/data C:\
scp -r -P 32222 capstone7@204.243.112.129:~/ELK/data ./

팁 ) SSH접속을 한다음 그안에서 명령어를 치면 ./부분이 로컬위치인데 로컬이 아니라 원격파일안위치에 복사가 된다. (원격->원격) 로컬 터미널에서 열고 명령어를 쳐야 제대로 복사가 되는것 같다.(원격->로컬)

<파일을 압축시켜서 원격으로 다운로드받기>

파일안에 대용량 파일들이 많을경우 속도가 너무 느려서 집파일을 만들어준다음 다운로드 해줬더니 그나마 빠르다.

1. ssh 원격서버에 접속한다음

2. 파일을 집파일로 만들어 저장해준다

tar -zcvf [파일명.tar.gz] [폴더명] 
#뒤 폴더명은 집파일로 만들 원래 폴더. 파일명은 새로 만들 파일명

3. 그리고 원격서버에서 생성된 집파일을 c드라이브안에 복사이동해준다

scp -P 32222 capstone7@204.243.112.129:~/datazip.tar.gz C:\

4. 아래그림처럼 다운로드가 진행된다.

1. ElasticSearch파일- bin - elasticsearch.bat 실행

2. kibana파일 - bin - kibana.bat 실행 - http://localhost:5601

3. logstash파일 - bin - logstash -f "C:\logstash-7.6.0\logstash-7.6.0\config\test2.conf" 

ELASTIC SEARCH명령어 윈도우용 - json 파일 업로드

curl -XPUT "http://localhost:9200/blog/user/dilbert" -H "Content-Type: application/json" -d "{ ""name"" : ""Dilbert Brown""}"" 
curl -XGET http://localhost:9200/blog?pretty 

curl -XPUT "http://localhost:9200/classes/class/1" -H "Content-Type: application/json" -d "{ ""title"" : ""Algorithm"",""professor"":""John""}" 
//document 추가 + 확인 
curl -XPOST "http://localhost:9200/classes/class/1" -H "Content-Type: application/json" _update -d "{ ""doc"" : {""unit"":1}}" 
curl -XGET http://localhost:9200/classes/class/1?pretty 
curl -XPOST "http://localhost:9200/classes/class/1" -H "Content-Type: application/json" _update -d "{ ""doc"" : {""unit"":2}}" 
//또다른방법  
curl -XPOST "http://localhost:9200/classes/class/1" -H "Content-Type: application/json" _update -d "{ ""script"" : ""ctx._source.unit+=5""}" 
//bulk파일 
curl -s -XPOST "http://localhost:9200/_bulk" -H "Content-Type: application/json" --data-binary @classes.json 
curl -XGET http://localhost:9200/classes?pretty 
//mapping ->이방식은 이미 생성된 index에 대한 내장필드를 설정하는방법이다. 
curl -XPUT "http://localhost:9200/yoonseo/class/_mapping?include_type_name=true" -d @classesRating_mapping.json -H "Content-Type: application/json" 
curl -XGET http://localhost:9200/yoonseo?pretty 
//데이터 넣어줌 
curl -XPOST "http://localhost:9200/yoonseo/_bulk"?pretty -H "Content-Type: application/json" --data-binary @classes.json 
curl -XGET http://localhost:9200/yoonseo/class/3?pretty 
//search 
curl -s -XPOST "http://localhost:9200/_bulk" -H "Content-Type: application/json" --data-binary @simple_basketball.json 
curl -XGET http://localhost:9200/basketball/record/_search?pretty 
검색1 
curl -XGET "localhost:9200/basketball/record/_search?q=points:30&pretty" 
검색2 //error 
curl -XGET "localhost:9200/basketball/record/_search -d" 
{   
   "query":{   
      "term":{ "points":30} 
   } 
}" 
//aggregation 
curl -XPOST "http://localhost:9200/_search"?pretty -H "Content-Type: application/json" --data-binary @avg_points_aggs.json 
curl -XPOST "http://localhost:9200/_search"?pretty -H "Content-Type: application/json" --data-binary @max_points_aggs.json 
curl -XPOST "http://localhost:9200/_search"?pretty -H "Content-Type: application/json" --data-binary @stats_points_aggs.json 

//bulk_aggregaion (basket지우고다시) 
curl -XPUT localhost:9200/basketball  
curl -XPUT "http://localhost:9200/basketball/record/_mapping?include_type_name=true" -d @basketball_mapping.json -H "Content-Type: application/json" 
curl -s -XPOST "http://localhost:9200/_bulk" --data-binary @bulk_basketball.json -H "Content-Type: application/json"  
curl -XGET http://localhost:9200/basketball/record/_search?pretty 
curl -XGET "http://localhost:9200/_search"?pretty -H "Content-Type: application/json" --data-binary @terms_aggs.json #에러 
curl -XGET "http://localhost:9200/_search"?pretty -H "Content-Type: application/json" --data-binary @stats_points_aggs.json 
curl -XGET "http://localhost:9200/_search?pretty --data-binary @terms_aggs.json 

curl -s -XPOST "http://localhost:9200/_bulk" -H "Content-Type: application/json" --data-binary @bulk_basketball.json 
curl -s -XPOST "http://localhost:9200/_bulk" -H "Content-Type: application/json" --data-binary @classes.json 

curl -X POST -H "Content-Type: application/x-ndjson" localhost:9200/_bulk --data-binary "@bulk_basketball.json" 
{  
   "size": 0, 
   "aggs": { // 집계  
      "my_aggs": { // 집계 명 (사용자 정의) 
         "{metric_arrgs_type}": { // 집계 타입  
            "field": "{file_name}" // 집계 대상 필드 
             "order" : { "{field_name}" : "desc" } // 정렬 
         } 
      } 
   } 
}  //매트릭집계구조 GET /{index_name}/_search?pretty 

logstash -  대량 csv파일 업로드용

---> 에러 발생시 logstash - logs안에 있는 제일 최근 텍스트 열어서 확인하기

아래 참조

새 창에서 열기

[Elastalert 설치 가이드라인]

해당 가이드라인은 ELK 스택 7.7버전을 사용하였습니다.

$ pip install elastalert
$ git clone https://github.com/Yelp/elastalert.git
$ pip install "setuptools>=11.3"
$ python setup.py install
$ elastalert-create-index

[Elastalert 실행 가이드라인]

• 테스트용으로 정상적으로 실행되는지만 확인
  $ elastalert-test-rule example_rules/d.yaml
• 실행
  $ elastalert --verbose --rule example_rules/d.yaml

(d.yaml대신 실제로 실행할 rule을 선택.)

[Elastalert Example_rules 가이드라인]

Rule타입

• Any : The any rule will match everything. Every hit that the query returns will generate an alert.
• Blacklist : The blacklist rule will check a certain field against a blacklist, and match if it is in the blacklist.
• Whitelist : Blacklist와 흡사 , this rule will compare a certain field to a whitelist, and match if the list does not contain the term.
• Change : This rule will monitor a certain field and match if that field changes.
• Frequency : This rule matches when there are at least a certain number of events in a given time frame.
• Flatline : This rule matches when the total number of events is under a given threshold for a time period.
• New Term :This rule matches when a new value appears in a field that has never been seen before
• Cardinality : This rule matches when a the total number of unique values for a certain field within a time frame is higher or lower than a threshold
• Metric Aggregation : This rule matches when the value of a metric within the calculation window is higher or lower than a threshold

samplerules.yamles_port: 14900 name: Example rule type: frequency index: logstash-* num_events: 50 timeframe: hours: 4 filter: - term: some_field: "some_value" alert: - "email" - "slack" email: - "bbbbbbbbb@naver.com" slack: slack_webhook_url: "https://hooks.slack.com/services/T~~~9B1OvKw" slack_username_override: 'wow159357' slack_channel_override: '#elastalert' slack_emoji_override: ':pizza:'

• es_host, es_port는 elasticsearch를 설치할떄의 ip,port번호를 적으면 된다.
• name 에 해당 룰 이름을 지정해준다.
• Type에 위에서 설명해준 Rule타입을 선택한다.
• time frame 해당 룰이 몇시간 이내에 실행된데이터를 기준으로 할지 정해준다.
• num event 에 특정데이터수를 지정해준다
• filter에 어떤 필드를 기준으로 삼을것인지 정해준다.
• alert에 'slack', 'email'등 알림받기 원하는것을 선택하여 적어준다.
• slack: 밑에 알람받을 슬랙주소와 유저주소, 채널,알람받을 이모티콘을 설정해준다.

[Elastalert Slack Webhook 설정]

1. 설정 - Add an app or custom integration 클릭
2. 가입된 이메일+비밀번호 입력
3. 슬렉웹페이지가 열리면 가운데 검색창에 incoming webhooks 선택
4. 왼쪽 메뉴바에 Add Configuration누르기
5. 메세지보낼 채널을 선택 (ex: #general)
6. webhook URL복사후 Rule에 들어가 Slak: 아래부분 slack_webhook_url: 에 작성

https://data.seoul.go.kr/dataList/datasetList.do

ㄴㄴ 여기서 각자 마음에 드는 데이터 하나골라서 csv 파일형식으로 다운받는다.

나는서울특별시 공공자전거 대여소별 이용정보(월별) 을 다운받았다.  (2018년도 csv파일 다운~)

데이터 다운받아서 열어보니까 이런형식이다.

(한글깨짐 현상을 방지하려면 CSV UTF-8(쉼표로분리)(*.csv)형식으로 새로 저장해주자.

원래 저장형식은 CSV(쉼표로분리)(*.csv) 이건데 이대로 쓰면 뒤에가서 업로드했을때 한글 깨진다.)

명심해야 할점 : 맨위에 필드들 ('대여일자', '대여소번호' 등등) 필드값이 있는 첫번쨰 행을 삭제해준다.!!!!!!

필드값은 다른곳에 적어놔서 꼭 순서대로 기억해둔다.

(왜냐햐면 로그스태시에서 1행도 데이터로 인식하기 때문에 맨윗행 삭제 안해주면 대여일자 : 대여일자 / 대여소번호 : 대여소번호 이렇게 잘못된 데이터 하나가 더 생긴다)

이걸토대로 logstash 에 sample.conf 파일을 생성해준다.

input {
        file {
                path => [ "C:/Users/Owner/Downloads/bicycle2.csv" ] 
	    start_position => "beginning"
	    sincedb_path => "nul"
	    codec => plain{ charset => "UTF-8" }
        }
}
filter {
        csv{ 
                columns => ["대여일자","대여소번호","대여소","대여건수","반납건수"]
                separator => ","
        }
}
output { 
        elasticsearch {
                hosts => ["localhost"] 
	   index => "softwarearchitecture"
        }
        stdout { } 
}

path에 데이터가 저장된 절대 경로 적어주고

colums에 각자 데이터 필드값을 적어준다.

index에는 Elasticsearch에 데이터 넣어둘 인덱스이름을 적어둔다.

(인덱스 없으면 새로만들어준다. 엘라스틱서치 실행시키고 아무 터미널이나 열고  curl -XPUT localhost:9200/softwarearchitecture 적어서 실행하면 된다. software~은 내가 정해준 인덱스이름)

logstash 파일안 bin 에 들어가서 터미널열고 실행 ( logstash -f (conf파일들어있는 절대경로) 실행해주면 된다. )

 logstash -f C:\logstash-7.6.0\logstash-7.6.0\config\sample.conf

잘 실행되었으면 위그림처럼 데이터들을 주르륵 보여진다.

(timestamp은 귀찮아서 설정하지 않았다. 설정하고싶다면 sample.conf에서 대여일자를 타임필터로 설정해주면 된다)

Kibana를 실행시키고 데이터가 잘들어갔나 확인해보자.

데이터가 잘 들어간것같다. 데이터가 제대로 들어간것을 확인하고 나서 대시보드를 취향껏 꾸며서 분석해준다.