(self, running_apps, addl_tags)
Get metrics for each application structured stream.
        Requires:
        - The Metric Servlet to be enabled to path <APP_URL>/metrics/json (enabled by default)
        - `SET spark.sql.streaming.metricsEnabled=true` in the app
Get metrics for each application structured stream.
        Requires:
        - The Metric Servlet to be enabled to path <APP_URL>/metrics/json (enabled by default)
        - `SET spark.sql.streaming.metricsEnabled=true` in the app
Get metrics for each application structured stream . Requires : - The Metric Servlet to be enabled to path <APP_URL > / metrics / json ( enabled by default ) - SET spark . sql . streaming . metricsEnabled = true in the app
def _spark_structured_streams_metrics(self, running_apps, addl_tags):
        """
        Get metrics for each application structured stream.
        Requires:
        - The Metric Servlet to be enabled to path <APP_URL>/metrics/json (enabled by default)
        - `SET spark.sql.streaming.metricsEnabled=true` in the app
        """

        for app_name, tracking_url in itervalues(running_apps):
            try:
                base_url = self._get_request_url(tracking_url)
                response = self._rest_request_to_json(
                    base_url, self.metricsservlet_path, SPARK_SERVICE_CHECK, addl_tags
                )
                self.log.debug('Structured streaming metrics: %s', response)
                response = {
                    metric_name: v['value']
                    for metric_name, v in iteritems(response.get('gauges'))
                    if 'streaming' in metric_name and 'value' in v
                }
                for gauge_name, value in iteritems(response):
                    match = STRUCTURED_STREAMS_METRICS_REGEX.match(gauge_name)
                    if not match:
                        self.log.debug("No regex match found for gauge: '%s'", str(gauge_name))
                        continue
                    groups = match.groupdict()
                    metric_name = groups['metric_name']
                    if metric_name not in SPARK_STRUCTURED_STREAMING_METRICS:
                        self.log.debug("Unknown metric_name encountered: '%s'", str(metric_name))
                        continue
                    metric_name, submission_type = SPARK_STRUCTURED_STREAMING_METRICS[metric_name]
                    tags = ['app_name:%s' % str(app_name)]
                    tags.extend(addl_tags)

                    if self._enable_query_name_tag:
                        query_name = groups['query_name']
                        tags.append('query_name:%s' % str(query_name))

                    self._set_metric(metric_name, submission_type, value, tags=tags)
            except HTTPError as e:
                self.log.debug(
                    "No structured streaming metrics to collect from" " app %s. %s", app_name, e, exc_info=True
                )
                pass
def _spark_structured_streams_metrics ( self , running_apps , addl_tags ) : for app_name , tracking_url in itervalues ( running_apps ) : try : base_url = self . _get_request_url ( tracking_url ) response = self . _rest_request_to_json ( base_url , self . metricsservlet_path , SPARK_SERVICE_CHECK , addl_tags ) self . log . debug ( 'Structured streaming metrics: %s' , response ) response = { metric_name : v [ 'value' ] for metric_name , v in iteritems ( response . get ( 'gauges' ) ) if 'streaming' in metric_name and 'value' in v } for gauge_name , value in iteritems ( response ) : match = STRUCTURED_STREAMS_METRICS_REGEX . match ( gauge_name ) if not match : self . log . debug ( "No regex match found for gauge: '%s'" , str ( gauge_name ) ) continue groups = match . groupdict ( ) metric_name = groups [ 'metric_name' ] if metric_name not in SPARK_STRUCTURED_STREAMING_METRICS : self . log . debug ( "Unknown metric_name encountered: '%s'" , str ( metric_name ) ) continue metric_name , submission_type = SPARK_STRUCTURED_STREAMING_METRICS [ metric_name ] tags = [ 'app_name:%s' % str ( app_name ) ] tags . extend ( addl_tags ) if self . _enable_query_name_tag : query_name = groups [ 'query_name' ] tags . append ( 'query_name:%s' % str ( query_name ) ) self . _set_metric ( metric_name , submission_type , value , tags = tags ) except HTTPError as e : self . log . debug ( "No structured streaming metrics to collect from" " app %s. %s" , app_name , e , exc_info = True ) pass