Finalize the toolsAPI helper methods and classes (#1893)

Signed-off-by: Ahmed Hussein (amahussein) <[email protected]>

Fixes #1887

Simplifies the interface in order to use the Tools-API more efficiently.

This pull request refactors the API v1 report handler system in the
RAPIDS PyTools codebase to simplify and modernize the way core report
handlers are created and used. The main change is the removal of the
legacy `APIHelpers` builder pattern in favor of direct class-based
instantiation for report handlers (e.g., `QualCore`, `ProfCore`). This
results in cleaner, more readable code and a more consistent interface
for handling reports. Additionally, the `CombinedDFBuilder` is replaced
with a new `CombinedCSVBuilder` class and metaclass for combining CSV
reports.

---------

Signed-off-by: Ahmed Hussein (amahussein) <[email protected]>

Author: amahussein

user_tools/src/spark_rapids_pytools/rapids/profiling.py

@@ -24,7 +24,6 @@
 from spark_rapids_pytools.common.sys_storage import FSUtil
 from spark_rapids_pytools.common.utilities import Utils
 from spark_rapids_pytools.rapids.profiling_core import ProfilingCore
-from spark_rapids_tools.api_v1.builder import TXTReport
 from spark_rapids_tools.utils.data_utils import TXTResult
 
 
@@ -130,7 +129,7 @@ def __generate_report_with_recommendations(self):
         sec_comments_head = ['\tComments:']
         log_lines.append('### Recommended configurations ###')
         headers = self.ctxt.get_value('local', 'output', 'summaryColumns')
-        with TXTReport(self.core_handler).table('appRawSummaryLog') as app_rawlogs_summaries:
+        with self.core_handler.txt('appRawSummaryLog') as app_rawlogs_summaries:
             # the result of the load operation is a dictionary [appid, data]
             for app_id, log_content in app_rawlogs_summaries.items():
                 app_name = ''

user_tools/src/spark_rapids_pytools/rapids/profiling_core.py

@@ -18,11 +18,11 @@
 from typing import List
 
 from spark_rapids_pytools.rapids.rapids_tool import RapidsJarTool
-from spark_rapids_tools.api_v1 import ProfCoreResultHandler
+from spark_rapids_tools.api_v1 import ProfCore
 
 
 @dataclass
-class ProfilingCore(RapidsJarTool[ProfCoreResultHandler]):
+class ProfilingCore(RapidsJarTool[ProfCore]):
     """
     Core profiling tool
     """

user_tools/src/spark_rapids_pytools/rapids/qualification.py

@@ -29,7 +29,6 @@
 from spark_rapids_pytools.common.sys_storage import FSUtil
 from spark_rapids_pytools.common.utilities import Utils, TemplateGenerator
 from spark_rapids_pytools.rapids.qualification_core import QualificationCore
-from spark_rapids_tools.api_v1 import APIHelpers, CSVReport
 from spark_rapids_tools.enums import QualFilterApp, QualEstimationModel, SubmissionMode
 from spark_rapids_tools.tools.additional_heuristics import AdditionalHeuristics
 from spark_rapids_tools.tools.cluster_config_recommender import ClusterConfigRecommender
@@ -418,7 +417,7 @@ def create_stdout_table_pprinter(total_apps: pd.DataFrame,
             return TopCandidates(props=view_dic, total_apps=total_apps, tools_processed_apps=tools_processed_apps)
 
         # 1. Read summary report using QualCoreHandler
-        with CSVReport(self.core_handler, _tbl='qualCoreCSVSummary') as q_sum_res:
+        with self.core_handler.csv('qualCoreCSVSummary') as q_sum_res:
             df = q_sum_res.data
         # 1. Operations related to XGboost modelling
         if not df.empty and self.ctxt.get_ctxt('estimationModelArgs')['xgboostEnabled']:
@@ -434,12 +433,9 @@ def create_stdout_table_pprinter(total_apps: pd.DataFrame,
                 ) from e
         # 2. Operations related to cluster information
         try:
-            with APIHelpers.CombinedDFBuilder(
-                    table='clusterInfoJSONReport',
-                    handlers=self.core_handler,
-                    raise_on_empty=False,
-                    raise_on_failure=False
-            ) as c_builder:
+            with self.core_handler.csv_combiner(
+                    'clusterInfoJSONReport'
+            ).suppress_failure() as c_builder:
                 # convert the json columns to csv columns
                 c_builder.apply_on_report(lambda x: x.map_cols(Qualification.__map_cluster_info_table()))
                 # use "App ID" included in the json report
@@ -455,17 +451,14 @@ def create_stdout_table_pprinter(total_apps: pd.DataFrame,
                               'Reason - %s:%s', type(e).__name__, e)
 
         # 3. Operations related to reading qualification output (unsupported operators and apps status)
-        with APIHelpers.CombinedDFBuilder(
-                table='unsupportedOpsCSVReport',
-                handlers=self.core_handler,
-                raise_on_empty=False,
-                raise_on_failure=False
-        ) as c_builder:
+        with self.core_handler.csv_combiner(
+                'unsupportedOpsCSVReport'
+        ).suppress_failure() as c_builder:
             # use "App ID" column name on the injected apps
             c_builder.combiner.on_app_fields({'app_id': 'App ID'})
             unsupported_ops_df = c_builder.build()
 
-        with CSVReport(self.core_handler, _tbl='coreCSVStatus') as status_res:
+        with self.core_handler.csv('coreCSVStatus') as status_res:
             apps_status_df = status_res.data
 
         # 4. Operations related to output
@@ -684,12 +677,9 @@ def __assign_spark_runtime_to_apps(self,
         Assigns the Spark Runtime (Spark/Photon) to each application. This will be used to categorize
         applications into speedup categories (Small/Medium/Large).
         """
-        with APIHelpers.CombinedDFBuilder(
-            table='coreRawApplicationInformationCSV',
-            handlers=self.core_handler,
-            raise_on_empty=False,
-            raise_on_failure=False
-        ) as c_builder:
+        with self.core_handler.csv_combiner(
+                'coreRawApplicationInformationCSV'
+        ).suppress_failure() as c_builder:
             # customize the report loading to only select required columns and rename them
             c_builder.apply_on_report(
                 lambda r: r.pd_args(

user_tools/src/spark_rapids_pytools/rapids/qualification_core.py

@@ -18,11 +18,11 @@
 from typing import List
 
 from spark_rapids_pytools.rapids.rapids_tool import RapidsJarTool
-from spark_rapids_tools.api_v1 import QualCoreResultHandler
+from spark_rapids_tools.api_v1 import QualCore
 
 
 @dataclass
-class QualificationCore(RapidsJarTool[QualCoreResultHandler]):
+class QualificationCore(RapidsJarTool[QualCore]):
     """
     Base class for qualification tools that provides core qualification functionality.
     """

user_tools/src/spark_rapids_pytools/rapids/qualification_stats.py

@@ -22,7 +22,7 @@
 from spark_rapids_pytools.common.utilities import Utils
 from spark_rapids_pytools.rapids.rapids_tool import RapidsTool
 from spark_rapids_pytools.rapids.tool_ctxt import ToolContext
-from spark_rapids_tools.api_v1 import APIHelpers
+from spark_rapids_tools.api_v1.builder import QualCore
 from spark_rapids_tools.tools.qualification_stats_report import SparkQualificationStats
 
 
@@ -74,8 +74,7 @@ def _process_output_args(self) -> None:
         self.ctxt.set_local('outputFolder', self.output_folder)
         self.logger.info('Local output folder is set as: %s', self.output_folder)
         # Add QualCoreHandler to the context
-        self.ctxt.set_ctxt('coreHandler',
-                           APIHelpers.QualCore.build_handler(dir_path=self.qual_output))
+        self.ctxt.set_ctxt('coreHandler', QualCore(self.qual_output))
 
     def _run_rapids_tool(self) -> None:
         """

user_tools/src/spark_rapids_pytools/rapids/qualx/prediction.py

@@ -18,7 +18,7 @@
 
 from spark_rapids_pytools.common.sys_storage import FSUtil
 from spark_rapids_pytools.rapids.qualx.qualx_tool import QualXTool
-from spark_rapids_tools.api_v1 import QualCoreResultHandler, APIHelpers
+from spark_rapids_tools.api_v1 import QualCore
 from spark_rapids_tools.tools.qualx.qualx_main import predict
 from spark_rapids_tools.tools.qualx.util import print_summary, print_speedup_summary
 
@@ -37,8 +37,8 @@ class Prediction(QualXTool):
     name = 'prediction'
 
     @property
-    def qual_handler(self) -> QualCoreResultHandler:
-        return APIHelpers.QualCore.build_handler(dir_path=self.qual_output)
+    def qual_handler(self) -> QualCore:
+        return QualCore(self.qual_output)
 
     def __prepare_prediction_output_info(self) -> dict:
         """

user_tools/src/spark_rapids_pytools/rapids/rapids_tool.py

@@ -37,7 +37,8 @@
 from spark_rapids_pytools.rapids.rapids_job import RapidsJobPropContainer
 from spark_rapids_pytools.rapids.tool_ctxt import ToolContext
 from spark_rapids_tools import CspEnv
-from spark_rapids_tools.api_v1 import ToolResultHandlerT, APIHelpers
+from spark_rapids_tools.api_v1 import ToolResultHandlerT
+from spark_rapids_tools.api_v1 import APIResHandler, QualCore, ProfCore
 from spark_rapids_tools.configuration.common import RuntimeDependency
 from spark_rapids_tools.configuration.submission.distributed_config import DistributedToolsConfig
 from spark_rapids_tools.configuration.tools_config import ToolsConfig
@@ -448,7 +449,7 @@ class RapidsJarTool(RapidsTool, Generic[ToolResultHandlerT]):
     """
 
     @cached_property
-    def core_handler(self) -> ToolResultHandlerT:
+    def core_handler(self) -> APIResHandler[ToolResultHandlerT]:
         """
         Create and return a coreHandler instance for reading core reports.
         This property should always be called after the scala code has executed.
@@ -459,9 +460,9 @@ def core_handler(self) -> ToolResultHandlerT:
         """
         normalized_tool_name = self.name.lower()
         if 'qualification' in normalized_tool_name:
-            return APIHelpers.QualCore.build_handler(dir_path=self.csp_output_path)
+            return QualCore(self.csp_output_path)
         if 'profiling' in normalized_tool_name:
-            return APIHelpers.ProfCore.build_handler(dir_path=self.csp_output_path)
+            return ProfCore(self.csp_output_path)
         raise ValueError(f'Tool name [{normalized_tool_name}] has no CoreHandler associated with it.')
 
     def _process_jar_arg(self):

user_tools/src/spark_rapids_tools/api_v1/__init__.py

@@ -42,7 +42,12 @@
 from .builder import (
     CSVReportCombiner,
     CSVReport,
-    APIHelpers
+    QualCore,
+    ProfCore,
+    QualWrapper,
+    ProfWrapper,
+    APIResHandler,
+    CombinedCSVBuilder
 )
 
 __all__ = [
@@ -60,5 +65,10 @@
     'ProfCoreResultHandler',
     'CSVReportCombiner',
     'CSVReport',
-    'APIHelpers'
+    'QualCore',
+    'ProfCore',
+    'QualWrapper',
+    'ProfWrapper',
+    'APIResHandler',
+    'CombinedCSVBuilder'
 ]