Function to transform log file to dataframe
(self, log_filepath)
| 31 | self.n_workers = n_workers |
| 32 | |
| 33 | def load_to_dataframe(self, log_filepath): |
| 34 | """Function to transform log file to dataframe""" |
| 35 | print("Loading log messages to dataframe...") |
| 36 | lines = [] |
| 37 | with open(log_filepath, "r") as fid: |
| 38 | lines = fid.readlines() |
| 39 | |
| 40 | log_messages = [] |
| 41 | if self.n_workers == 1: |
| 42 | log_messages = formalize_message(enumerate(lines), self.regex, self.headers) |
| 43 | else: |
| 44 | chunk_size = np.ceil(len(lines) / float(self.n_workers)) |
| 45 | chunks = groupby( |
| 46 | enumerate(lines), key=lambda k, line=count(): next(line) // chunk_size |
| 47 | ) |
| 48 | log_chunks = [list(chunk) for _, chunk in chunks] |
| 49 | print("Read %d log chunks in parallel" % len(log_chunks)) |
| 50 | pool = mp.Pool(processes=self.n_workers) |
| 51 | result_chunks = [ |
| 52 | pool.apply_async( |
| 53 | formalize_message, args=(chunk, self.regex, self.headers) |
| 54 | ) |
| 55 | for chunk in log_chunks |
| 56 | ] |
| 57 | pool.close() |
| 58 | pool.join() |
| 59 | log_messages = list(chain(*[result.get() for result in result_chunks])) |
| 60 | |
| 61 | if not log_messages: |
| 62 | raise RuntimeError("Logformat error or log file is empty!") |
| 63 | log_dataframe = pd.DataFrame(log_messages, columns=["LineId"] + self.headers) |
| 64 | success_rate = len(log_messages) / float(len(lines)) |
| 65 | print( |
| 66 | "Loading {} messages done, loading rate: {:.1%}".format( |
| 67 | len(log_messages), success_rate |
| 68 | ) |
| 69 | ) |
| 70 | return log_dataframe |
| 71 | |
| 72 | def _generate_logformat_regex(self, logformat): |
| 73 | """Function to generate regular expression to split log messages""" |
no test coverage detected