freeEDU/Log-Decoder
0
1"""2The interface to load log datasets. The datasets currently supported include3HDFS and BGL.4 5Authors:6 LogPAI Team7 8"""9 10import pandas as pd11import os12import numpy as np13import re14from sklearn.utils import shuffle15from collections import OrderedDict16 17def _split_data(x_data, y_data=None, train_ratio=0, split_type='uniform'):18 if split_type == 'uniform' and y_data is not None:19 pos_idx = y_data > 020 x_pos = x_data[pos_idx]21 y_pos = y_data[pos_idx]22 x_neg = x_data[~pos_idx]23 y_neg = y_data[~pos_idx]24 train_pos = int(train_ratio * x_pos.shape[0])25 train_neg = int(train_ratio * x_neg.shape[0])26 x_train = np.hstack([x_pos[0:train_pos], x_neg[0:train_neg]])27 y_train = np.hstack([y_pos[0:train_pos], y_neg[0:train_neg]])28 x_test = np.hstack([x_pos[train_pos:], x_neg[train_neg:]])29 y_test = np.hstack([y_pos[train_pos:], y_neg[train_neg:]])30 elif split_type == 'sequential':31 num_train = int(train_ratio * x_data.shape[0])32 x_train = x_data[0:num_train]33 x_test = x_data[num_train:]34 if y_data is None:35 y_train = None36 y_test = None37 else:38 y_train = y_data[0:num_train]39 y_test = y_data[num_train:]40 # Random shuffle41 indexes = shuffle(np.arange(x_train.shape[0]))42 x_train = x_train[indexes]43 if y_train is not None:44 y_train = y_train[indexes]45 return (x_train, y_train), (x_test, y_test)46 47def load_HDFS(log_file, label_file=None, window='session', train_ratio=0.5, split_type='sequential', save_csv=False, window_size=0):48 """ Load HDFS structured log into train and test data49 50 Arguments51 ---------52 log_file: str, the file path of structured log.53 label_file: str, the file path of anomaly labels, None for unlabeled data54 window: str, the window options including `session` (default).55 train_ratio: float, the ratio of training data for train/test split.56 split_type: `uniform` or `sequential`, which determines how to split dataset. `uniform` means57 to split positive samples and negative samples equally when setting label_file. `sequential`58 means to split the data sequentially without label_file. That is, the first part is for training,59 while the second part is for testing.60 61 Returns62 -------63 (x_train, y_train): the training data64 (x_test, y_test): the testing data65 """66 67 print('====== Input data summary ======')68 69 if log_file.endswith('.npz'):70 # Split training and validation set in a class-uniform way71 data = np.load(log_file)72 x_data = data['x_data']73 y_data = data['y_data']74 (x_train, y_train), (x_test, y_test) = _split_data(x_data, y_data, train_ratio, split_type)75 76 elif log_file.endswith('.csv'):77 assert window == 'session', "Only window=session is supported for HDFS dataset."78 print("Loading", log_file)79 struct_log = pd.read_csv(log_file, engine='c',80 na_filter=False, memory_map=True)81 data_dict = OrderedDict()82 for idx, row in struct_log.iterrows():83 blkId_list = re.findall(r'(blk_-?\d+)', row['Content'])84 blkId_set = set(blkId_list)85 for blk_Id in blkId_set:86 if not blk_Id in data_dict:87 data_dict[blk_Id] = []88 data_dict[blk_Id].append(row['EventId'])89 data_df = pd.DataFrame(list(data_dict.items()), columns=['BlockId', 'EventSequence'])90 91 if label_file:92 # Split training and validation set in a class-uniform way93 label_data = pd.read_csv(label_file, engine='c', na_filter=False, memory_map=True)94 label_data = label_data.set_index('BlockId')95 label_dict = label_data['Label'].to_dict()96 data_df['Label'] = data_df['BlockId'].apply(lambda x: 1 if label_dict[x] == 'Anomaly' else 0)97 98 # Split train and test data99 (x_train, y_train), (x_test, y_test) = _split_data(data_df['EventSequence'].values, 100 data_df['Label'].values, train_ratio, split_type)101 102 print(y_train.sum(), y_test.sum())103 104 if save_csv:105 data_df.to_csv('data_instances.csv', index=False)106 107 if window_size > 0:108 x_train, window_y_train, y_train = slice_hdfs(x_train, y_train, window_size)109 x_test, window_y_test, y_test = slice_hdfs(x_test, y_test, window_size)110 log = "{} {} windows ({}/{} anomaly), {}/{} normal"111 print(log.format("Train:", x_train.shape[0], y_train.sum(), y_train.shape[0], (1-y_train).sum(), y_train.shape[0]))112 print(log.format("Test:", x_test.shape[0], y_test.sum(), y_test.shape[0], (1-y_test).sum(), y_test.shape[0]))113 return (x_train, window_y_train, y_train), (x_test, window_y_test, y_test)114 115 if label_file is None:116 if split_type == 'uniform':117 split_type = 'sequential'118 print('Warning: Only split_type=sequential is supported \119 if label_file=None.'.format(split_type))120 # Split training and validation set sequentially121 x_data = data_df['EventSequence'].values122 (x_train, _), (x_test, _) = _split_data(x_data, train_ratio=train_ratio, split_type=split_type)123 print('Total: {} instances, train: {} instances, test: {} instances'.format(124 x_data.shape[0], x_train.shape[0], x_test.shape[0]))125 return (x_train, None), (x_test, None), data_df126 else:127 raise NotImplementedError('load_HDFS() only support csv and npz files!')128 129 num_train = x_train.shape[0]130 num_test = x_test.shape[0]131 num_total = num_train + num_test132 num_train_pos = sum(y_train)133 num_test_pos = sum(y_test)134 num_pos = num_train_pos + num_test_pos135 136 print('Total: {} instances, {} anomaly, {} normal' \137 .format(num_total, num_pos, num_total - num_pos))138 print('Train: {} instances, {} anomaly, {} normal' \139 .format(num_train, num_train_pos, num_train - num_train_pos))140 print('Test: {} instances, {} anomaly, {} normal\n' \141 .format(num_test, num_test_pos, num_test - num_test_pos))142 143 return (x_train, y_train), (x_test, y_test)144 145def slice_hdfs(x, y, window_size):146 results_data = []147 print("Slicing {} sessions, with window {}".format(x.shape[0], window_size))148 for idx, sequence in enumerate(x):149 seqlen = len(sequence)150 i = 0151 while (i + window_size) < seqlen:152 slice = sequence[i: i + window_size]153 results_data.append([idx, slice, sequence[i + window_size], y[idx]])154 i += 1155 else:156 slice = sequence[i: i + window_size]157 slice += ["#Pad"] * (window_size - len(slice))158 results_data.append([idx, slice, "#Pad", y[idx]])159 results_df = pd.DataFrame(results_data, columns=["SessionId", "EventSequence", "Label", "SessionLabel"])160 print("Slicing done, {} windows generated".format(results_df.shape[0]))161 return results_df[["SessionId", "EventSequence"]], results_df["Label"], results_df["SessionLabel"]162 163 164 165def load_BGL(log_file, label_file=None, window='sliding', time_interval=60, stepping_size=60, 166 train_ratio=0.8):167 """ TODO168 169 """170 171 172def bgl_preprocess_data(para, raw_data, event_mapping_data):173 """ split logs into sliding windows, built an event count matrix and get the corresponding label174 175 Args:176 --------177 para: the parameters dictionary178 raw_data: list of (label, time)179 event_mapping_data: a list of event index, where each row index indicates a corresponding log180 181 Returns:182 --------183 event_count_matrix: event count matrix, where each row is an instance (log sequence vector)184 labels: a list of labels, 1 represents anomaly185 """186 187 # create the directory for saving the sliding windows (start_index, end_index), which can be directly loaded in future running188 if not os.path.exists(para['save_path']):189 os.mkdir(para['save_path'])190 log_size = raw_data.shape[0]191 sliding_file_path = para['save_path']+'sliding_'+str(para['window_size'])+'h_'+str(para['step_size'])+'h.csv'192 193 #=============divide into sliding windows=========#194 start_end_index_list = [] # list of tuples, tuple contains two number, which represent the start and end of sliding time window195 label_data, time_data = raw_data[:,0], raw_data[:, 1]196 if not os.path.exists(sliding_file_path):197 # split into sliding window198 start_time = time_data[0]199 start_index = 0200 end_index = 0201 202 # get the first start, end index, end time203 for cur_time in time_data:204 if cur_time < start_time + para['window_size']*3600:205 end_index += 1206 end_time = cur_time207 else:208 start_end_pair=tuple((start_index,end_index))209 start_end_index_list.append(start_end_pair)210 break211 # move the start and end index until next sliding window212 while end_index < log_size:213 start_time = start_time + para['step_size']*3600214 end_time = end_time + para['step_size']*3600215 for i in range(start_index,end_index):216 if time_data[i] < start_time:217 i+=1218 else:219 break220 for j in range(end_index, log_size):221 if time_data[j] < end_time:222 j+=1223 else:224 break225 start_index = i226 end_index = j227 start_end_pair = tuple((start_index, end_index))228 start_end_index_list.append(start_end_pair)229 inst_number = len(start_end_index_list)230 print('there are %d instances (sliding windows) in this dataset\n'%inst_number)231 np.savetxt(sliding_file_path,start_end_index_list,delimiter=',',fmt='%d')232 else:233 print('Loading start_end_index_list from file')234 start_end_index_list = pd.read_csv(sliding_file_path, header=None).values235 inst_number = len(start_end_index_list)236 print('there are %d instances (sliding windows) in this dataset' % inst_number)237 238 # get all the log indexes in each time window by ranging from start_index to end_index239 expanded_indexes_list=[]240 for t in range(inst_number):241 index_list = []242 expanded_indexes_list.append(index_list)243 for i in range(inst_number):244 start_index = start_end_index_list[i][0]245 end_index = start_end_index_list[i][1]246 for l in range(start_index, end_index):247 expanded_indexes_list[i].append(l)248 249 event_mapping_data = [row[0] for row in event_mapping_data]250 event_num = len(list(set(event_mapping_data)))251 print('There are %d log events'%event_num)252 253 #=============get labels and event count of each sliding window =========#254 labels = []255 event_count_matrix = np.zeros((inst_number,event_num))256 for j in range(inst_number):257 label = 0 #0 represent success, 1 represent failure258 for k in expanded_indexes_list[j]:259 event_index = event_mapping_data[k]260 event_count_matrix[j, event_index] += 1261 if label_data[k]:262 label = 1263 continue264 labels.append(label)265 assert inst_number == len(labels)266 print("Among all instances, %d are anomalies"%sum(labels))267 assert event_count_matrix.shape[0] == len(labels)268 return event_count_matrix, labels269