Updated

Author: syncmachineuser

parsing_models/application_model.py

@@ -6,6 +6,7 @@
 import os
 import boto3
 import gzip
+from pprint import pprint
 
 from .job_model import JobModel
 from .executor_model import ExecutorModel
@@ -51,6 +52,11 @@ def __init__(self, eventlogpath, bucket=None, stdoutpath=None):
 
         self.shuffle_partitions = 200
 
+
+        self.cloud_platform = None
+        self.cloud_provider = None
+        self.spark_version = None
+
         # if bucket is None, then files are in local directory, else read from s3
         # read event log
         if bucket is None:
@@ -95,8 +101,9 @@ def __init__(self, eventlogpath, bucket=None, stdoutpath=None):
                 json_data = get_json(line)
                 event_type = json_data["Event"]
                 if event_type == "SparkListenerLogStart":
-                    spark_version_dict = {"spark_version": json_data["Spark Version"]}
-                    self.spark_metadata = {**self.spark_metadata, **spark_version_dict}
+                    #spark_version_dict = {"spark_version": json_data["Spark Version"]}
+                    self.spark_version = json_data["Spark Version"]
+                    self.spark_metadata = {**self.spark_metadata}
                 elif event_type == "SparkListenerJobStart":
 
                     job_id = json_data["Job ID"]
@@ -134,8 +141,9 @@ def __init__(self, eventlogpath, bucket=None, stdoutpath=None):
                 elif event_type == "SparkListenerStageCompleted":
 
                     # stages may not be executed exclusively from one job
-                    self.finish_time = json_data['Stage Info']['Completion Time']/1000
                     stage_id = json_data['Stage Info']["Stage ID"]
+                    self.finish_time = json_data['Stage Info']['Completion Time']/1000
+
                     for job_id in self.jobs_for_stage[stage_id]:
                         self.jobs[job_id].stages[stage_id].completion_time = json_data['Stage Info']['Completion Time']/1000
 
@@ -146,8 +154,10 @@ def __init__(self, eventlogpath, bucket=None, stdoutpath=None):
 
                     # This if is specifically for databricks logs
                     if 'spark.databricks.clusterUsageTags.sparkVersion' in curKeys:
-                        json_data['Spark Properties']['platform'] = 'Databricks'
-                        json_data['Spark Properties']['spark_version'] = json_data['Spark Properties']['spark.databricks.clusterUsageTags.sparkVersion']
+
+                        self.cloud_platform = 'databricks'
+                        self.cloud_provider = json_data['Spark Properties']['spark.databricks.clusterUsageTags.cloudProvider'].lower()
+                        self.spark_version = json_data['Spark Properties']['spark.databricks.clusterUsageTags.sparkVersion']
 
                     self.spark_metadata = {**self.spark_metadata, **json_data["Spark Properties"]}
 
@@ -234,8 +244,11 @@ def __init__(self, eventlogpath, bucket=None, stdoutpath=None):
                 self.jobs[0].add_event(line, False)
 
 
-        if 'platform' not in self.spark_metadata.keys():
-            self.spark_metadata['platform'] = 'EMR'
+        if self.cloud_platform == None:
+            self.cloud_platform = 'emr'
+            self.cloud_provider = 'aws'
+            #self.spark_metadata['cloud_platform'] = 'emr'
+            #self.spark_metadata['cloud_provider'] = 'aws'
 
         self.dag.decipher_dag()
         self.dag.add_broadcast_dependencies(self.stdoutpath)

parsing_models/application_model_v2.py

@@ -11,22 +11,19 @@
 import boto3
 from collections import defaultdict
 
-logging.basicConfig(format='%(levelname)s:%(message)s')
-
 class sparkApplication():
 
     def __init__(self, 
         objfile  = None, # Previously saved object. This is the fastest and best option
         appobj   = None, # application_model object
         eventlog = None, # spark eventlog path,
         stdout   = None,
-        debug    = False
         ):
 
         self.eventlog = eventlog
         self.existsSQL = False
         self.existsExecutors = False
-        self.sparkMetadata = {}
+        #self.sparkMetadata = {}
         self.metadata = {}
         self.stdout = stdout
 
@@ -157,13 +154,13 @@ def getExecutorInfo(self, appobj):
         df = defaultdict(lambda: [])
         for xid, executor in appobj.executors.items():
 
-            #print(executor.end_time)
+            # print(executor.end_time)
             # Special case for handling end_time
             if executor.end_time is not None:
                 end_time = executor.end_time/1000 - appobj.start_time
             else:
-                #print('None detected')
-                end_time = appobj.finish_time - appobj.start_time
+                # print('None detected')
+                end_time = executor.end_time
 
             df['executor_id'].append(xid)
             df['cores']      .append(executor.cores)
@@ -336,7 +333,9 @@ def getAllTaskData(self, appobj):
             'end_time'      : end_time,
             'duration'      : duration,
             #'input_mb'      : input_mb,
-            'remote_mb_read': remote_mb_read,
+
+            # Duplicate entry:
+            # 'remote_mb_read': remote_mb_read,
             'locality'      : locality,
 
             # Disk-based performance metrics
@@ -560,11 +559,24 @@ def getAllDriverAccumData(self, appobj):
         self.accumData = df  
 
     def getAllMetaData(self, appobj):
-        self.sparkMetadata = (appobj.spark_metadata)
-        self.metadata = {"app_name": appobj.app_name,
-                         "start_time": appobj.start_time}
-
 
+        #self.sparkMetadata = (appobj.spark_metadata)
+
+        self.metadata = {
+            'application_info' : {
+                'timestamp_start_ms' : int(appobj.start_time*1000),
+                'timestamp_end_ms' : int(appobj.finish_time*1000),
+                'runtime_sec' : appobj.finish_time - appobj.start_time,
+                'name' : appobj.app_name,
+                'id' : appobj.spark_metadata['spark.app.id'],
+                'spark_version' : appobj.spark_version,
+                'cloud_platform' : appobj.cloud_platform,
+                'cloud_provider' : appobj.cloud_provider
+
+            },
+            'spark_params' : appobj.spark_metadata
+        }
+    
     def addMetadata(self, key=None, value=None):
 
         if (key is None) or (value is None):
@@ -640,7 +652,7 @@ def save(self, filepath=None, compress=False):
             saveDat['executors'] = self.executorData.reset_index().to_dict('list')
 
         saveDat['metadata'] = self.metadata
-        saveDat['sparkMetadata'] = self.sparkMetadata
+        #saveDat['sparkMetadata'] = self.sparkMetadata
         saveDat['metadata']['existsSQL']       = self.existsSQL
         saveDat['metadata']['existsExecutors'] = self.existsExecutors 
 
@@ -703,21 +715,36 @@ def load(self, filepath=None):
         self.metadata        = saveDat['metadata']
         self.existsSQL       = self.metadata.pop('existsSQL')
         self.existsExecutors = self.metadata.pop('existsExecutors')
-        self.sparkMetadata   = saveDat.pop('sparkMetadata')
+
+        # This is for legacy support and should be removed after it is in production for a few
+        # weeks. Introduced 3/9/2022 by SDG.
+        if 'sparkMetadata' in saveDat:
+            self.sparkMetadata   = saveDat.pop('sparkMetadata')
 
+        # SPC113 - SDG
+        # Because of the way jobData is created, if there are no job Events in the eventlog then the
+        # correct fields will not exist. A second condition checking for the 'job_id' field is 
+        # necessary here to ensure this method will run if this is the case.
+        #
+        # Note: stageData is initialized differently so this same issue does not exist for that
+        # structure. Furthermore, in the event that 'jobData' has no values within, 'stageData' will
+        # also have no values and an invalidLog exception will be thrown during log validation
+        # in SparkApplicaionAdvanced.
+        if ('jobData' in saveDat) and ('job_id' in saveDat['jobData']):
+            self.jobData = pd.DataFrame.from_dict(saveDat['jobData'])
+            self.jobData = self.jobData.set_index('job_id')
 
-        if 'jobData' in saveDat:   self.jobData   = pd.DataFrame.from_dict(saveDat['jobData']  ).set_index('job_id')
         if 'stageData' in saveDat: self.stageData = pd.DataFrame.from_dict(saveDat['stageData']).set_index('stage_id')
         if 'taskData' in saveDat:  self.taskData  = pd.DataFrame.from_dict(saveDat['taskData'] ).set_index('task_id')
-        if 'accumData' in saveDat:  
+        if 'accumData' in saveDat:
             self.accumData  = pd.DataFrame.from_dict(saveDat['accumData'] )
             if 'sql_id' in self.accumData.columns:
                 self.accumData = self.accumData.set_index('sql_id')
 
         if self.existsSQL:
             self.sqlData      = pd.DataFrame.from_dict(saveDat['sqlData']).set_index('sql_id')
         if self.existsExecutors:
-            self.executorData = pd.DataFrame.from_dict(saveDat['executors']).set_index('executor_id')  
+            self.executorData = pd.DataFrame.from_dict(saveDat['executors']).set_index('executor_id')
 
         logging.info('Loaded object from: %s [%.2f]' % (filepath, (time.time()-t1)))
 

parsing_models/stage_model.py

@@ -1,8 +1,4 @@
-import collections
-import json
 import numpy
-import matplotlib.pyplot as plt
-import seaborn as sns
 
 from .task_model import TaskModel
 

parsing_models/utility.py

@@ -1,35 +1,3 @@
-import re
-
-def convert_to_MiB(size_string):
-    """
-    Function to convert to data sizes to MiB from string 
-    e.g. "10g", "10 GB"
-    """
-    # default if no units mentioned is MiB
-    if size_string.isdigit():
-        data_size = float(size_string)
-    else:
-        pattern = "([0-9]+) *([a-zA-z]+)"
-        match = re.match(pattern, size_string.strip(' '))
-
-        data_size = float(match.group(1))
-        unit = match.group(2)
-
-        if unit == 'kiB':
-            data_size = data_size / 1024
-        elif unit == 'GiB' or unit == 'g':
-            data_size = data_size * 1024
-        elif unit == 'GB':        
-            data_size = data_size * 10**9 / 1024 / 1024
-        elif unit == 'MB':        
-            data_size = data_size * 10**6 / 1024 / 1024        
-        elif unit == 'kB':
-            data_size = data_size * 10**3 / 1024 / 1024
-        elif unit == 'bytes' or unit == 'B':
-            data_size = data_size / 1024 / 1024
-    return data_size
-
-
 def db_to_aws_configs(appobj):
     meta = appobj.sparkMetadata
     rt = 'spark.databricks.clusterUsageTags.'

spark_predictor

@@ -1 +1 @@
-Subproject commit 22fb7c178388a5302fbb6d29545171cd963b9237
+Subproject commit d09b5326da36328df30159a00de696b05f5e8b59