Pandas dataframe export

README.md

@@ -52,6 +52,7 @@ acelyzer -i "${TRACE_DIR}/hap-json-files/hap-bs8-seq256-autopilot-0-34707-job-*.
 
 An evolving feature is the use of processing profiles (option `-P`) to allow control which processing stages are enabled. By default, the `everything.json` profile (or `default.json`) is used. Note that the cmdline still overrides the deactivation of stages so that if a stage is not requested via cmdline, its activation in the profile has no effect. When creating a profile, it's currently necessary to start from the everything-profile and set unwanted stages to `false`. If the provided profile file is not found at the given path, the tool will lookup a file with that name at the location of the other predefined profiles (`<install_or_src_path>/src/aiu_trace_analyzer/profiles`).
 
+
 ### Input Files
 
 The tool is capable of ingesting these types of input files:
@@ -62,7 +63,6 @@ The tool is capable of ingesting these types of input files:
 There's a basic autodetect function for the file type built in. If the filename extension doesn't indicate the type, it detects log files by their lines with time stamps, it detects json files by finding the initial open parenthesis, and for everything else, it assumes binary format of perfetto.
 
 
-
 ### Output Files
 
 The outputs from running Acelyzer include several files, providing insights into AIU performance and
@@ -81,6 +81,9 @@ the stdout log files associated with a profiling run). For example, the `bmm*` c
 operations using the PT-arrays in AIU’s Rapid-Cores, while other categories are primarily responsible
 for vector computations that do not involve the PT-array.
 
+ * Pandas dataframe export (`-f pddf`): Allows [integration](#integration-into-other-tools) into other post processing scripts like Jupyter notebooks. If the output to file is not disabled, the data is also exported as a text file table.
+
+
 **Table: Available Performance Metrics/Views**
 | Metrics/Features | Description | Details |
 |------------------|-------------|---------|
@@ -94,6 +97,7 @@ for vector computations that do not involve the PT-array.
 | Kernel time breakdown | Function profile of execution time and occurrences of device-residing computation kernels | Available in text and csv format for autopilot-off profiling mode |
 | Elapsed time per inference iteration | Averaged elapsed time of an inference iteration | Available in console output for autopilot-off profiling mode, only if autodetection succeeded |
 
+
 ### Important Kernel Profile Statistics
 
 Acelyzer creates basic kernel profile statistics in the output file named `<name>_categories.{txt,csv}`. It's created in txt and csv format for consumption by either human inspection or subsequent processing respectively.
@@ -121,28 +125,34 @@ There are 3 types of time stamps synchronization involved:
 
 When running `acelyzer`, the progress and basic processing information is printed to the console. This includes information about the input/output files and events as well as essential warnings and errors. The event processing is happening in stages and each stage has a short identifier like OVC (overflow correction) or PEC (power extraction and computation), etc. This helps to identify problems and categorize any warnings.  It is recommended to pay some attention to the warnings before diving into further analysis or visualization because they can point to certain problems with the input or output data which may render useless the visualized data. For example if the tool detects an effective frequency that wildly differs from what's used on the cmdline, your event durations and timestamps could be unrealistic/unreliable. More about this can be found in the [troubleshooting](#understanding-and-troubleshooting-the-results) section.
 
+
 ## Options/Event Processing in Detail
 
 This section explains some of the command line options in more detail.
 
+
 ### compiler_log
 This option allows to pass the compiler log output into `acelyzer` and allows for additional data augmentation and PT-Array utilization data. For multi-AIU workloads, one file per process should be provided to avoid intermingled data inside the log. Also it is important that multi-AIU logfiles be provided in the order of the process ranks to properly map each log file to its rank.
 
+
 ### flow
 For multi-AIU workloads, this option enables a detection of communication calls and collective operations.
 
 Limitations:
  * It strongly depends on the available flex data and might just not be able to find communication primitives or the necessary information to correlate sends and receives to establish flow events.
  * If collective events are not detected, it also prevents computation of collective bandwidth and the distributed view data in TensorBoard.
 
+
 ### build_coll_event
 
 This feature relies on successful flow detection to create an event stream where each event spans the duration of collective operations and allows for computation of effective bandwidth for these collectives. This can be useful for analysis of distributed workloads.
 
+
 ### comm_summarize_seq
 
 Somewhat similar to [build_coll_event](#build_coll_event), but doesn't require flow detection. It creates a communication event for each detected pair of send-receive events. This is useful to increase the visual comprehension of communication steps and reduce the amount of events for cases where a single send or receive operation consists of multiple sub-steps.
 
+
 ### flex_ts_fix
 Acelyzer is attempting to detect cycle-to-wallclock mapping differences between jobs of the input. There are 2 problems that it tries to address:
  1) if the AIU events are shifted outside of the corresponding AIU-Roundtrip event: In this case, there's a per-job (and rank) offset that's calculated between the first AIU event and the AIU-Roundtrip event.
@@ -161,6 +171,7 @@ Consequences when in use:
  * the same offset is applied to all events of a job
  * the amount of adjustment is added to the event metadata, so have immediate information that the event was shifted and by how much.
 
+
 ### comm_summarize_seq
 
 This option replaces the individual events of a Flex communication with a single combined send or recv event spanning the time of the separate items. While this removes detail from the result, it reduces 'clutter' and allows better visual clarity to follow a communication protocol at a higher level.
@@ -169,6 +180,7 @@ Consequences when in use:
  * removes detailed sub-steps that show when and how interactions with the host appear
  * spans the entire time from posting of the communication to the completion, this can exaggerate communication time e.g. if a recv is posted early for later completion
 
+
 ### freq
 
 This option allows to set the SoC and Ideal frequencies if they differ from the default.
@@ -213,11 +225,11 @@ Consequences when in use:
  * nothing negative other than the increased number of events
 
 
-
 ## Understanding and Troubleshooting the Results
 
 Sometimes the input data results in strange results. Some cases are being discussed below:
 
+
 ### Misaligned Events: with Gap
 
 This can happen when flex events within a job have hit a cycle counter overflow and the calibration was unable to compute the correct wallclock timestamps.
@@ -255,6 +267,7 @@ What can be done:
  * the option `--flex_ts_fix` should be able to alleviate this problem (see [here](#flex_ts_fix) for details and consequences).
  * worst case: rerunning the experiment can help
 
+
 ### Misaligned Event: Stretched
 
 This can happen if the 'effective' SoC frequency doesn't fit the actual frequency. The AIU Flex events are based on cycle counters and the conversion from cycles to wallclock is based on a set frequency.
@@ -294,6 +307,7 @@ What can be done:
 
 The output json file can be viewed in chrome, perfetto, or tensorboard.
 
+
 ### Chrome
 
  * Open url `chrome://tracing` and click the `load` button (or drag and drop the json file onto the tracing window)
@@ -318,6 +332,33 @@ The command line option `--tb` enables additional post-processing steps for bett
 Go to `https://ui.perfetto.dev/` and select `open trace file` (or drag and drop the json file onto the tracing window). Note that this is an online service even if the current claim is that everything runs locally in your browser.
 
 
+### Integration Into Other Tools
+
+The cmdline args `-i api://jsonbuffer` and/or `--disable_file` allow to inject and extract data via memory instead of files. This enables integration with other post processing scripts or tools. Below is an example for use in Jupyter notebooks using Pandas dataframe export format:
+
+```
+from aiu_trace_analyzer.core.acelyzer import Acelyzer
+
+# adjust pointing to a trace data base-directory
+TRACE_BASE = "../data/traces"
+# adjust to pick a sub-directory of the base dir
+TRACE_NAME = "mymodel_trace"
+# define the output format (pddf=pandas.dataframe, json=json string)
+output_format = "pddf"
+
+# acelyzer args
+args = [
+    "-i", f"{TRACE_BASE}/{TRACE_NAME}/*rank_*-of_4-job-[345].json",
+    "-c", f"{TRACE_BASE}/{TRACE_NAME}/compiler-log.txt",
+    "--disable_file",
+    "-f", output_format]
+
+# create, run, and extract the result trace data
+ace = Acelyzer(args)
+ace.run()
+trace_data = ace.get_output_data()
+```
+
 
 ## Developer Info
 

src/aiu_trace_analyzer/core/acelyzer.py

@@ -17,6 +17,13 @@
 from aiu_trace_analyzer import __version__
 
 
+class AcelyzerArgsFormatter(argparse.RawTextHelpFormatter, argparse.ArgumentDefaultsHelpFormatter):
+    """
+    Combine argsparse formatting for preserved line breaks and including default values
+    """
+    pass
+
+
 class Acelyzer:
 
     defaults = {
@@ -156,9 +163,18 @@ def run(self) -> int:
                 self.exporter = output.JsonFileTraceExporter(target_uri=self.args.output,
                                                              timescale=self.args.time_unit,
                                                              settings=vars(self.args))
-            else:
+            elif self.args.format == "pddf":
+                self.exporter = output.DataframeExporter(target_uri=self.args.output,
+                                                         timescale=self.args.time_unit,
+                                                         settings=vars(self.args))
+            elif self.args.format == "proto":
                 self.exporter = output.ProtobufTraceExporter(target_uri=self.args.output,
                                                              settings=vars(self.args))
+            else:
+                aiulog.log(
+                    aiulog.ERROR,
+                    "Unrecognized export format. Available options: json or pddf (and unsupported: proto)")
+                return -1
         self.exporter.export_meta(importer.get_passthrough_meta())
 
         self.register_processing_functions(process, self.args, self.exporter)
@@ -172,7 +188,7 @@ def run(self) -> int:
 
     def parse_inputs(self, args=None):
         # to include default value in --help output
-        parser = argparse.ArgumentParser(prog="acelyzer", formatter_class=argparse.RawTextHelpFormatter)
+        parser = argparse.ArgumentParser(prog="acelyzer", formatter_class=AcelyzerArgsFormatter)
         required_group = parser.add_mutually_exclusive_group(required=True)
         parser.add_argument("-C", "--counter", type=str, nargs='*', default=self.defaults["counter"],
                             choices=["power_ts4", "power_ts3", "coll_bw", "bandwidth", "prep_queue", "rcu_util"],
@@ -199,7 +215,8 @@ def parse_inputs(self, args=None):
                             help="List of event types to keep. E.g. 'C' to just keep counters.")
 
         parser.add_argument("-f", "--format", type=str, default=self.defaults["format"],
-                            help="Type of output format (json, protobuf)")
+                            choices=["json", "pddf", "protobuf"],
+                            help="Type of output format")
 
         parser.add_argument("--freq", type=str, default=':'.join([str(self.defaults["freq"]),
                                                                   str(self.defaults["ideal_freq"])]),
@@ -336,11 +353,8 @@ def parse_inputs(self, args=None):
                 parsed_args.profile = self.defaults["stage_profile"]
         return parsed_args
 
-    def get_output_data(self) -> str:
-        if self.args.tb and self.args.tb_refinement:
-            return self.exporter.get_data()
-        else:
-            return "Only supported for TensorBoard exporter."
+    def get_output_data(self):
+        return self.exporter.get_data()
 
     def _args_sanity_check(self, args) -> bool:
         assert math.isclose(args.freq_scaling, 0.0, abs_tol=1e-9), \

src/aiu_trace_analyzer/export/exporter.py

@@ -3,6 +3,9 @@
 import sys
 import os
 from collections import defaultdict
+from typing import Optional
+
+import pandas as pd
 
 import aiu_trace_analyzer.logger as aiulog
 import aiu_trace_analyzer.trace_view as tv
@@ -52,6 +55,9 @@ def export(self, _data: list[tv.AbstractEventType]):
     def flush(self):
         raise NotImplementedError("Class %s doesn't implement flush()" % (self.__class__.__name__))
 
+    def get_data(self):
+        raise NotImplementedError("Class %s doesn't implement get_data()" % (self.__class__.__name__))
+
 
 class JsonFileTraceExporter(AbstractTraceExporter):
     '''
@@ -74,6 +80,10 @@ def export_meta(self, meta_data):
     def export_raw(self, data: dict):
         self.traceview.append_trace_event(data)
 
+    # return traceview data as a json string
+    def get_data(self) -> str:
+        return self.traceview.dump(fp=None)
+
     # write the traceview to file
     def flush(self):
         assert isinstance(self.device_data, list)
@@ -169,9 +179,6 @@ def _save_overall_trace(self) -> None:
             with open(file_name, 'w') as json_new_pids_file:
                 self.traceview.dump(fp=json_new_pids_file)
 
-    def get_data(self) -> str:
-        return self.traceview.dump(fp=None)
-
     def get_tb_data(self, worker) -> str:
         return self.traceview_by_rank[worker].dump(fp=None)
 
@@ -204,3 +211,81 @@ def flush(self):
             return
 
         self._save_events_by_id()
+
+
+class DataframeExporter(AbstractTraceExporter):
+    def __init__(
+            self, target_uri, timescale="ms", settings=None,
+            data_map: dict = None):
+        super().__init__(target_uri=target_uri, settings=settings)
+        self.vertical_view = []
+        self.df = None
+
+        # mapping from event entry to dataframe column
+        # only entries that appear are picked up
+        if not data_map:
+            self.data_map = {
+                "ts": ("Timestamp", 0.0),
+                "dur": ("Duration", 0.0),
+                "cat": ("Category", "other"),
+                "name": ("Event Name", "NoName"),
+                "args.pt_active": ("PT_Active", 0.0)}
+        else:
+            self.data_map = data_map
+
+    def add_device(self, id, data: dict):
+        devdata = {"id": id}
+        for k, v in data.items():
+            devdata[k] = v
+        self.device_data.append(devdata)
+
+        assert isinstance(self.device_data, list)
+
+    def export_meta(self, meta_data: dict) -> None:
+        # no metadata for this exporter type
+        return
+
+    def _extract_value(self, key_path: str, event: dict) -> str:
+        try:
+            default = self.data_map[key_path][1]
+        except KeyError:
+            return "N/A"
+
+        keys = key_path.split('.')
+        value = event
+        for k in keys:
+            if isinstance(value, dict) and k in value:
+                value = value[k]
+            else:
+                return default
+        return value
+
+    def _convert_trace_event(self, event_dict: tv.AbstractEventType) -> Optional[tuple]:
+        if event_dict.ph != "X":
+            return None
+
+        rval = []
+        for jpath in self.data_map.keys():
+            rval.append(self._extract_value(jpath, event_dict.json()))
+
+        return tuple(rval)
+
+    # export (a list) of events to the configured target
+    def export(self, data: list[tv.AbstractEventType]):
+        for event in data:
+            if not isinstance(event, tv.CompleteEvents):
+                continue
+
+            event_line = self._convert_trace_event(event)
+            if event_line:
+                self.vertical_view.append(event_line)
+
+    def flush(self):
+        title_row = [v[0] for v in self.data_map.values()]
+        self.df = pd.DataFrame(self.vertical_view, columns=title_row)
+
+        with open(self.target_uri, 'w') as f:
+            f.write(self.df.to_string(index=False))
+
+    def get_data(self) -> pd.DataFrame:
+        return self.df

src/aiu_trace_analyzer/pipeline/normalize.py

@@ -137,7 +137,10 @@ def extract_eventfilters(self, filterstr: str) -> dict[str, re.Pattern]:
                 aiulog.log(aiulog.WARN, "FLTR: key:regex pattern not found in event filter. Skipping", fstr)
                 continue
             event_filters[key_regex[0]] = re.compile(rf"{key_regex[1]}")
-        aiulog.log(aiulog.INFO, f"FLTR: Event filtering is active. {len(event_filters)} filters enabled.")
+        aiulog.log(
+            aiulog.INFO,
+            f"FLTR: Event filtering is active. {len(event_filters)} filters enabled:",
+            event_filters)
         return event_filters
 
     def event_filtered(self, event: TraceEvent) -> bool:

src/aiu_trace_analyzer/pipeline/rcu_utilization.py

@@ -730,6 +730,8 @@ def compute_utilization(event: TraceEvent, context: AbstractContext) -> list[Tra
                 cmpt_dur,
                 job_fingerprint)
         util_counter = context.make_utilization_event(event, utilization*100.0)
+        if utilization > 0.0:
+            event["args"]["pt_active"] = utilization
         return [event] + util_counter
 
     return [event]