· 8 years ago · Dec 05, 2017, 12:36 PM
1from Utils import JSONFile, Hadoop
2from Utils.LogManager import LogManager
3from Utils.LogManager import logger
4from Main.WorkspaceManager import WorkspaceManager
5from MongoDBManager import MongoDBManager
6from YarnManager import YarnManager
7from SparkManager import SparkManager
8import ProjectFunction
9import sys
10import os
11from Main.SparkJob import SparkJob
12from Main.Pipeline import Pipeline
13from multiprocessing import Process, Pipe
14import traceback
15from Model.ModellingManager import ModellingManager
16from datetime import datetime
17
18class Project(object):
19 """
20 Handles the work flow of project, and contains project related functions.
21
22 @note: We can create multiple instances of SparkJob, to submit jobs to Spark.
23 Each SparkJob has a semaphore which limit the number of jobs you can submit through to Spark
24
25 @note: An environment has its own SparkJob object, which is used to submit jobs to Spark
26 Environment is a loosely defined term. It can be a main module (e.g. ETL, general feature engineering),
27 or it can be a modelling problem (merchant, period)
28 """
29
30 def __init__(self, sys_setup_json, crash_flag):
31 """
32 Constructor
33 Create a new Project Object.
34
35 @param crash_flag: True means Project object has crashed before and this is a restart
36 """
37
38 # contains the json data for sys_setup.json
39 self.setup_project(sys_setup_json, crash_flag)
40
41 def start(self):
42 """
43 Runs the main stages of the project
44 """
45 self.run_etl()
46 self.run_general_feature_eng()
47 self.run_model()
48
49
50 def run_model(self):
51 """
52 Read the modelling_problems_setup and run the modelling stage of the project
53 """
54 if 'modelling_problems_setup' not in self._workspace_manager.get_sys_var('sys_setup_json'):
55 logger.log(__name__ + '.' + sys._getframe().f_code.co_name + "Project setup file \
56 ('model_setup_file') is not found in setup JSON file", logger.WARNING)
57 return
58
59 modelling_problems_setup_file = self._workspace_manager.get_sys_var('project_local_path') \
60 + self._workspace_manager.get_sys_var('sys_setup_json')['modelling_problems_setup']
61
62 logger.log(__name__ + '.' + sys._getframe().f_code.co_name + " Loading Modelling setup \
63 from '" + modelling_problems_setup_file + "'...", logger.INFO)
64 modelling_problems_setup_file_json = JSONFile.read_file(modelling_problems_setup_file)
65
66 # replace json values
67 #JSONFile.replace_value_with_key(self._workspace_manager.get_sys_var('global_key'),
68 # modelling_problems_setup_file_json, '$global_key$')
69
70 mm = ModellingManager(self._workspace_manager)
71 mm.start(modelling_problems_setup_file_json)
72
73
74 def setup_project(self, sys_setup_json, crash_flag):
75 """
76 Start the process of setting up the project based on the JSON file sys_setup_json.
77
78 @note: for mongodb, project_id is used as the database name,
79 and each workspace has its own collection
80 """
81
82 (project_id, project_local_path, hadoop_bin_path, max_concurrent_jobs_for_a_sparkJob_obj,
83 hdfs_fs_defaultFS, relative_logging_path, logging_level, mongodb_host, debug_mode,
84 local_text_path, spark_log_path_in_hdfs, completed_pipeline_list_setting,
85 hadoop_host, hadoop_user, max_concurrent_processes, spark_shuffle_partitions,
86 mongodb_user, mongodb_password)\
87 = ProjectFunction.setup_project_json_check(sys_setup_json)
88
89 # 1. create logger
90 log_path = project_local_path + relative_logging_path
91 logger = LogManager(log_path)
92 logger.set_logging_level(logging_level)
93 logger.create_logger(str(project_id))
94 logger.log("---------- Start project '" + project_id + ' ---------- ', logger.INFO)
95
96 # 2. setup mongodb manager
97 mongodb_manager = MongoDBManager(host=mongodb_host)
98 # use project id as the db, and each workspace has its own collection
99 mongodb_manager.create_select_db(project_id, mongodb_user, mongodb_password)
100
101 # 3. setup yarn manager
102 yarn_manager = YarnManager(project_id, hadoop_bin_path, hdfs_fs_defaultFS)
103
104 # 4. setup local text path
105 if not os.path.exists(project_local_path + local_text_path):
106 #os.mkdir(project_local_path + local_text_path)
107 os.makedirs(project_local_path + local_text_path)
108
109 # 5. setup spark logging path in hadoop
110 yarn_manager.create_hdfs_path(spark_log_path_in_hdfs, delete_existing=False)
111
112 # 6. store global keys
113 #global_key_file = project_local_path + sys_setup_json['global_key_file']
114 global_key_file = project_local_path + "json/global_key.json"
115 global_key = JSONFile.read_file(global_key_file)
116
117 # 6. load workspace setup data. first load workspace setup JSON file
118 # deprecated it. We fixed it to json/workspaces.json
119 """
120 if 'workspace_setup_file' not in sys_setup_json:
121 raise KeyError("Project Workspace setup file('workspace_setup_file') is not found in \
122 setup JSON file....")
123 """
124 #workspace_setup_file = project_local_path + sys_setup_json['workspace_setup_file']
125 workspace_setup_file = project_local_path + "json/workspaces.json"
126 workspace_setup_data = JSONFile.read_file(workspace_setup_file)
127
128 # 7. prepare dict of system variable to store
129 # max_concurrent_jobs: limit the number of concurrent SparkContext in a SparkJob.
130 # Each SparkContext runs a pipeline
131 # max_concurrent_processes: limit the number of concurrent processes running jobs.
132 # Each job is taken from the job table
133 sys_var_dict = {
134 'sys_setup_json': sys_setup_json,
135 'debug_mode': debug_mode,
136 'max_concurrent_jobs_for_a_sparkJob_obj': max_concurrent_jobs_for_a_sparkJob_obj,
137 'max_concurrent_processes': max_concurrent_processes,
138 'project_id': project_id,
139 'project_local_path': project_local_path,
140 'global_key': global_key,
141 'local_text_path': local_text_path,
142 'completed_pipeline_list_setting': completed_pipeline_list_setting,
143 'hadoop_user': hadoop_user,
144 'hadoop_host': hadoop_host,
145 'spark_shuffle_partitions': spark_shuffle_partitions
146 }
147
148 self._workspace_manager = WorkspaceManager(project_id, yarn_manager, mongodb_manager,
149 sys_var_dict)
150
151 # 8. setup workspace in hdfs. Workspace is under project path in hdfs
152 self._workspace_manager.setup_workspace(workspace_setup_data)
153
154 # 9. Lastly, setup completed_pipeline_list
155 if completed_pipeline_list_setting['goto_last_processed_pipeline'] == 1:
156 self.prepare_completed_pipeline_list(completed_pipeline_list_setting, crash_flag)
157
158 # 10. delete all lock documents across mongo collections
159 self._workspace_manager.\
160 delete_json_dict_from_mongo_across_collections({'name': {'$regex': 'lock$'}})
161
162
163 def get_workspace_manager(self):
164 return self._workspace_manager
165
166 def prepare_completed_pipeline_list(self, completed_pipeline_list_setting, crash_flag):
167 """
168 completed_pipeline_list stores the completed pipelines of the project.
169 This is useful when we want to resume to the last pipeline before the system crashes.
170 Hence, we do not need to re-process the processed pipelines again
171
172 completed_pipeline_list is pickled on disk
173
174 We get the settings for completed_pipeline_list by two ways:
175 1. directly from JSON file completed_pipeline_list_setting
176 2 from "pipelines" key value pairs in JSON file (deprecated)
177
178
179 @param completed_pipeline_list_setting: dict containing the settings
180 If 'pipelines' is not in completed_pipeline_list_setting, we use
181
182 Pipeline.get_completed_pipeline_list_mode()
183
184 to decide how to handle completed pipelines
185
186 Pipeline.get_completed_pipeline_list_mode()
187 0: Do not record processed pipeline.
188 1: Clear existing records and start recording.
189 2: Start recording
190 3: completed_pipeline_list is corrupted, restore to backup.
191
192 Deprecated mode
193 11: Clear existing records and start recording. If program crashed and restarted,
194 existing records will not be cleared and the program will resume at where it crashed."
195 """
196 pl = Pipeline(self._workspace_manager)
197
198 if not os.path.exists(pl.get_completed_pipeline_list_path()):
199 os.makedirs(pl.get_completed_pipeline_list_path())
200
201 # if pipelines exists, use it to override other settings
202 if 'pipelines' in completed_pipeline_list_setting:
203 # do not use completed_pipeline_list in this case
204 pl.set_goto_last_processed_pipeline(0)
205
206 src_workspace_id = 'status'
207
208 # Prepare to get modelling_periods, put entities inside json
209 #replace_lists = #[(self._workspace_manager.get_sys_var('global_key'), "$global_key$")]
210
211 pipeline_json_wrapper = \
212 JSONFile.prepare_json_for_pipeline(completed_pipeline_list_setting, None,
213 src_workspace_id)
214
215 # set flag to get result_dict from spawn processes
216 kwargs = {'return_var': 1}
217 # suppose to return result_dict['status']
218 result_dict = pl.start(pipeline_json_wrapper, **kwargs)
219 # if status is 1, need to resume last crashed pipeline
220 if result_dict['status'] == 1:
221 # backup current version before start of this project
222 pl.backup_completed_pipeline_list()
223 elif result_dict['status'] == 0:
224 # if status is 0, clear history, start from beginning
225 pl.clear_completed_pipeline_list()
226 else:
227 raise ValueError('status %s not recognized' %(str(result_dict['status'])))
228
229 # update completed_pipeline_list_setting in workspace_manager
230 completed_pipeline_list_setting['goto_last_processed_pipeline'] = 1
231 completed_pipeline_list_setting['completed_pipeline_list_mode'] = 2
232 self._workspace_manager.store_sys_var(completed_pipeline_list_setting)
233 else:
234 # use the settings defined in json
235 if pl.get_completed_pipeline_list_mode() == 1:
236 pl.clear_completed_pipeline_list()
237 logger.log(__name__ + '.' + sys._getframe().f_code.co_name +
238 " Clear completed pipeline list and exit system", logger.INFO)
239 os._exit(0)
240 elif pl.get_completed_pipeline_list_mode() == 3:
241 # completed_pipeline_list is corrupted, restore to backup
242 pl.restore_completed_pipeline_list()
243 logger.log(__name__ + '.' + sys._getframe().f_code.co_name +
244 " Restore corrupted completed pipeline list and exit system",
245 logger.INFO)
246 os._exit(0)
247 elif pl.get_completed_pipeline_list_mode() == 11:
248 # Clear existing records and start recording. If program crashed and restarted,
249 # existing records will not be cleared and the program will resume at where it
250 # crashed.
251 if crash_flag == False:
252 pl.clear_completed_pipeline_list()
253 else:
254 # backup current version before start of this project
255 pl.backup_completed_pipeline_list()
256 elif pl.get_completed_pipeline_list_mode() == 4:
257 # Delete history before a given date
258 if 'datetime_threshold' not in completed_pipeline_list_setting:
259 raise KeyError('datetime_threshold not in %s'
260 %(str(completed_pipeline_list_setting)))
261 dt = completed_pipeline_list_setting['datetime_threshold']
262
263 if 'datetime_format' not in completed_pipeline_list_setting:
264 raise KeyError('datetime_format not in %s'
265 %(str(completed_pipeline_list_setting)))
266 df = completed_pipeline_list_setting['datetime_format']
267
268 datetime_threshold = datetime.strptime(dt, df)
269 pl.clear_completed_pipeline_list(datetime_threshold)
270 logger.log(__name__ + '.' + sys._getframe().f_code.co_name +
271 " Clear completed pipelines whose timestamps are earlier than %s" %(df),
272 logger.INFO)
273 os._exit(0)
274 else:
275 # backup current version before start of this project
276 pl.backup_completed_pipeline_list()
277
278 def run_etl(self):
279 """
280 Read the etl_setup_file and run the etl stage of the project
281 """
282 if 'etl_setup_file' not in self._workspace_manager.get_sys_var('sys_setup_json'):
283 logger.log(__name__ + '.' + sys._getframe().f_code.co_name +
284 " etl_setup_file not found. Exit ETL",
285 logger.WARNING)
286 return
287
288 etl_setup_file = self._workspace_manager.get_sys_var('project_local_path') + \
289 self._workspace_manager.get_sys_var('sys_setup_json')['etl_setup_file']
290
291 logger.log(__name__ + '.' + sys._getframe().f_code.co_name +
292 " Loading ETL tasks from '" + etl_setup_file + "'...", logger.INFO)
293 etl_setup_json = JSONFile.read_file(etl_setup_file)
294
295 # replace json values
296 #JSONFile.replace_value_with_key(self._workspace_manager.get_sys_var('global_key'),
297 # etl_setup_json, '$global_key$')
298 #JSONFile.replace_value(None, self._workspace_manager.get_sys_var('global_key'),
299 # etl_setup_json)
300
301 pl = Pipeline(self._workspace_manager)
302 pl.start(etl_setup_json)
303
304
305
306 def run_general_feature_eng(self):
307 """
308 Read the general_feat_eng_setup_file and run the general feature engineering stage of the
309 project
310 """
311
312 if 'general_feat_eng_setup_file' not in \
313 self._workspace_manager.get_sys_var('sys_setup_json'):
314 logger.log(__name__ + '.' + sys._getframe().f_code.co_name +
315 " general_feat_eng_setup_file not found. Exit General Feature Engineering",
316 logger.WARNING)
317 return 0
318
319 general_feat_eng_setup_file = self._workspace_manager.get_sys_var('project_local_path') + \
320 self._workspace_manager.get_sys_var('sys_setup_json')['general_feat_eng_setup_file']
321
322 logger.log(__name__ + '.' + sys._getframe().f_code.co_name +
323 " Loading General Feature Engineering tasks from '" + \
324 general_feat_eng_setup_file + "'...", logger.INFO)
325 general_feat_eng_setup_json = JSONFile.read_file(general_feat_eng_setup_file)
326
327 # replace json values
328 #JSONFile.replace_value_with_key(self._workspace_manager.get_sys_var('global_key'),
329 # general_feat_eng_setup_json, '$global_key$')
330
331 #JSONFile.replace_value(None, self._workspace_manager.get_sys_var('global_key'),
332 # general_feat_eng_setup_json)
333
334 pl = Pipeline(self._workspace_manager)
335 pl.start(general_feat_eng_setup_json)
336
337 return 1