spark_plugin.py
def _get_stage_metrics(self, apps):
        """
        Get stage metrics corresponding to a running application

        Args:
        apps (dict): Mapping of application id to app name, user,
                     and url endpoint for API call
        """
        for app_id, (app_name, app_user, tracking_url) in apps.items():
            resp = self.spark_agent.request_metrics(tracking_url, APPS_ENDPOINT, app_id, "stages", status="ACTIVE")
            if not resp:
                continue

            dim = {"app_name": str(app_name), "user": str(app_user)}
            dim.update(self.global_dimensions)

            stage_metrics = self.metrics[(app_name, app_user)]
            if len(resp):
                stage_metrics["spark.num_active_stages"] = MetricRecord(
                    "spark.num_active_stages", "gauge", len(resp), dim
                )

            for stage in resp:
                for key, (metric_type, metric_name) in metrics.SPARK_STAGE_METRICS.items():

                    if key not in stage:
                        continue

                    metric_value = stage[key]

                    mr = stage_metrics.get(key, MetricRecord(metric_name, metric_type, 0, dim))
                    mr.value += metric_value
                    stage_metrics[key] = mr

            for key, mr in stage_metrics.items():
                if key == "executorRunTime" and len(resp) > 0:
                    mr.value /= len(resp)

            if not self.enhanced_flag and len(self.include) == 0:
                continue

            for stage in resp:
                for key, (metric_type, metric_name) in metrics.SPARK_STAGE_METRICS_ENHANCED.items():

                    if key not in stage:
                        continue
                    if metric_name in self.exclude:
                        continue
                    if len(self.include) > 0 and metric_name not in self.include:
                        continue

                    metric_value = stage[key]

                    mr = stage_metrics.get(key, MetricRecord(metric_name, metric_type, 0, dim))
                    mr.value += metric_value
                    stage_metrics[key] = mr

        Get stage metrics corresponding to a running application

        Args:
        apps (dict): Mapping of application id to app name, user,
                     and url endpoint for API call
        
Get stage metrics corresponding to a running application
Args:
apps (dict): Mapping of application id to app name, user,
and url endpoint for API call
Get stage metrics corresponding to a running application Args : apps ( dict ) : Mapping of application id to app name user and url endpoint for API call
def _get_stage_metrics(self, apps):
        for app_id, (app_name, app_user, tracking_url) in apps.items():
            resp = self.spark_agent.request_metrics(tracking_url, APPS_ENDPOINT, app_id, "stages", status="ACTIVE")
            if not resp:
                continue
            dim = {"app_name": str(app_name), "user": str(app_user)}
            dim.update(self.global_dimensions)
            stage_metrics = self.metrics[(app_name, app_user)]
            if len(resp):
                stage_metrics["spark.num_active_stages"] = MetricRecord(
                    "spark.num_active_stages", "gauge", len(resp), dim
                )
            for stage in resp:
                for key, (metric_type, metric_name) in metrics.SPARK_STAGE_METRICS.items():
                    if key not in stage:
                        continue
                    metric_value = stage[key]
                    mr = stage_metrics.get(key, MetricRecord(metric_name, metric_type, 0, dim))
                    mr.value += metric_value
                    stage_metrics[key] = mr
            for key, mr in stage_metrics.items():
                if key == "executorRunTime" and len(resp) > 0:
                    mr.value /= len(resp)
            if not self.enhanced_flag and len(self.include) == 0:
                continue
            for stage in resp:
                for key, (metric_type, metric_name) in metrics.SPARK_STAGE_METRICS_ENHANCED.items():
                    if key not in stage:
                        continue
                    if metric_name in self.exclude:
                        continue
                    if len(self.include) > 0 and metric_name not in self.include:
                        continue
                    metric_value = stage[key]
                    mr = stage_metrics.get(key, MetricRecord(metric_name, metric_type, 0, dim))
                    mr.value += metric_value
                    stage_metrics[key] = mr
def _get_stage_metrics ( self , apps ) : for app_id , ( app_name , app_user , tracking_url ) in apps . items ( ) : resp = self . spark_agent . request_metrics ( tracking_url , APPS_ENDPOINT , app_id , "stages" , status = "ACTIVE" ) if not resp : continue dim = { "app_name" : str ( app_name ) , "user" : str ( app_user ) } dim . update ( self . global_dimensions ) stage_metrics = self . metrics [ ( app_name , app_user ) ] if len ( resp ) : stage_metrics [ "spark.num_active_stages" ] = MetricRecord ( "spark.num_active_stages" , "gauge" , len ( resp ) , dim ) for stage in resp : for key , ( metric_type , metric_name ) in metrics . SPARK_STAGE_METRICS . items ( ) : if key not in stage : continue metric_value = stage [ key ] mr = stage_metrics . get ( key , MetricRecord ( metric_name , metric_type , 0 , dim ) ) mr . value += metric_value stage_metrics [ key ] = mr for key , mr in stage_metrics . items ( ) : if key == "executorRunTime" and len ( resp ) > 0 : mr . value /= len ( resp ) if not self . enhanced_flag and len ( self . include ) == 0 : continue for stage in resp : for key , ( metric_type , metric_name ) in metrics . SPARK_STAGE_METRICS_ENHANCED . items ( ) : if key not in stage : continue if metric_name in self . exclude : continue if len ( self . include ) > 0 and metric_name not in self . include : continue metric_value = stage [ key ] mr = stage_metrics . get ( key , MetricRecord ( metric_name , metric_type , 0 , dim ) ) mr . value += metric_value stage_metrics [ key ] = mr
Get stage metrics corresponding to a running application
Args:
apps (dict): Mapping of application id to app name, user,
and url endpoint for API call
Get stage metrics corresponding to a running application Args : apps ( dict ) : Mapping of application id to app name user and url endpoint for API call
"""
        Get stage metrics corresponding to a running application

        Args:
        apps (dict): Mapping of application id to app name, user,
                     and url endpoint for API call
        """
self
apps