NDS-H support for warmup iterations (#211)

* Changes for removing duplicate PysparkBenchReport + nds-h warmup+iterations changes

* Changes for removing duplicate PysparkBenchReport + nds-h warmup+iterations changes

* Refactor for nds/nds-h

Signed-off-by: Sayed Bilal Bari <[email protected]>

* License updated

Signed-off-by: Sayed Bilal Bari <[email protected]>

* Consolidating common jvm_listener and aqe.properties

Signed-off-by: Sayed Bilal Bari <[email protected]>

* Updating license header

Signed-off-by: Sayed Bilal Bari <[email protected]>

* Removing old files and adding symlinks to new files

* Changes to test with symlinks

* Reverting NDS related changes

Signed-off-by: Sayed Bilal Bari <[email protected]>

* Reverting NDS related changes

Signed-off-by: Sayed Bilal Bari <[email protected]>

* Updated headers for files without any change

---------

Signed-off-by: Sayed Bilal Bari <[email protected]>

Author: sayedbilalbari

nds-h/nds_h_power.py

@@ -186,6 +186,8 @@ def run_query_stream(input_prefix,
                      query_dict,
                      time_log_output_path,
                      sub_queries,
+                     warmup_iterations,
+                     iterations,
                      input_format,
                      output_path=None,
                      keep_sc=False,
@@ -238,6 +240,8 @@ def run_query_stream(input_prefix,
         print("====== Run {} ======".format(query_name))
         q_report = PysparkBenchReport(spark_session, query_name)
         summary = q_report.report_on(run_one_query,
+                                     warmup_iterations,
+                                     iterations,
                                      spark_session,
                                      q_content,
                                      query_name,
@@ -347,13 +351,23 @@ def load_properties(filename):
                         default='parquet')
     parser.add_argument('--property_file',
                         help='property file for Spark configuration.')
+    parser.add_argument('--warmup_iterations',
+                        type=int,
+                        help='Number of warmup iterations for each query.',
+                        default=0)
+    parser.add_argument('--iterations',
+                        type=int,
+                        help='Number of iterations for each query.',
+                        default=1)
     args = parser.parse_args()
     query_dict = gen_sql_from_stream(args.query_stream_file)
     run_query_stream(args.input_prefix,
                      args.property_file,
                      query_dict,
                      args.time_log,
                      args.sub_queries,
+                     args.warmup_iterations,
+                     args.iterations,
                      args.input_format,
                      args.output_prefix,
                      args.keep_sc,

utils/python_benchmark_reporter/PysparkBenchReport.py

@@ -56,13 +56,14 @@ def __init__(self, spark_session: SparkSession, query_name) -> None:
             'query': query_name,
         }
 
-    def report_on(self, fn: Callable, *args):
+    def report_on(self, fn: Callable, warmup_iterations = 0, iterations = 1, *args):
         """Record a function for its running environment, running status etc. and exclude sentive
         information like tokens, secret and password Generate summary in dict format for it.
 
         Args:
             fn (Callable): a function to be recorded
-
+            :param iterations:
+            :param warmup_iterations:
         Returns:
             dict: summary of the fn
         """
@@ -83,28 +84,41 @@ def report_on(self, fn: Callable, *args):
         if listener is not None:
             print("TaskFailureListener is registered.")
         try:
-            start_time = int(time.time() * 1000)
-            fn(*args)
-            end_time = int(time.time() * 1000)
-            if listener and len(listener.failures) != 0:
-                self.summary['queryStatus'].append("CompletedWithTaskFailures")
-            else:
-                self.summary['queryStatus'].append("Completed")
+            # warmup
+            for i in range(0, warmup_iterations):
+                fn(*args)
         except Exception as e:
-            # print the exception to ease debugging
-            print('ERROR BEGIN')
+            print('ERROR WHILE WARMUP BEGIN')
             print(e)
             traceback.print_tb(e.__traceback__)
-            print('ERROR END')
-            end_time = int(time.time() * 1000)
-            self.summary['queryStatus'].append("Failed")
-            self.summary['exceptions'].append(str(e))
-        finally:
-            self.summary['startTime'] = start_time
-            self.summary['queryTimes'].append(end_time - start_time)
-            if listener is not None:
-                listener.unregister()
-            return self.summary
+            print('ERROR WHILE WARMUP END')
+
+        start_time = int(time.time() * 1000)
+        self.summary['startTime'] = start_time
+        # run the query
+        for i in range(0, iterations):
+            try:
+                start_time = int(time.time() * 1000)
+                fn(*args)
+                end_time = int(time.time() * 1000)
+                if listener and len(listener.failures) != 0:
+                    self.summary['queryStatus'].append("CompletedWithTaskFailures")
+                else:
+                    self.summary['queryStatus'].append("Completed")
+            except Exception as e:
+                # print the exception to ease debugging
+                print('ERROR BEGIN')
+                print(e)
+                traceback.print_tb(e.__traceback__)
+                print('ERROR END')
+                end_time = int(time.time() * 1000)
+                self.summary['queryStatus'].append("Failed")
+                self.summary['exceptions'].append(str(e))
+            finally:
+                self.summary['queryTimes'].append(end_time - start_time)
+        if listener is not None:
+            listener.unregister()
+        return self.summary
 
     def write_summary(self, prefix=""):
         """_summary_