__init__.py (11750B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\fsd\mapReduce\__init__.py 2 from __future__ import absolute_import 3 import os 4 import sys 5 import logging 6 import traceback 7 import time 8 from . import jobGenerator 9 import itertools 10 log = logging.getLogger(__name__) 11 log.addHandler(logging.NullHandler()) 12 13 def AbsJoin(*args): 14 return os.path.abspath(os.path.join(*args)) 15 16 17 class RemoteProcessException(Exception): 18 pass 19 20 21 class benchmark(object): 22 23 def __init__(self, name): 24 self.name = name 25 26 def __enter__(self): 27 self.start = time.time() 28 29 def __exit__(self, ty, val, tb): 30 end = time.time() 31 log.info('%s : %0.3f seconds' % (self.name, end - self.start)) 32 return False 33 34 35 def GetSimpleTaskGenerator(walker, chunkSize, directoryFiles, include, exclude, mapReduceFinalizeFunctions): 36 taskID = 0 37 for chunk in ChunkIterable(jobGenerator.GetAllMatchingFiles(walker, directoryFiles, include, exclude), chunkSize): 38 taskCount = 0 39 task = RemoteProcessTask(taskID, walker) 40 taskID += 1 41 for filePath, relativePath, directoryPathStackForFile in chunk: 42 validMapReduceOperations = [ job.GetValidMapReduceOperations(relativePath) for job in mapReduceFinalizeFunctions ] 43 count = sum(map(len, validMapReduceOperations)) 44 taskCount += count 45 if count: 46 task.AddFileToTask(relativePath, directoryPathStackForFile, validMapReduceOperations) 47 48 yield task 49 50 51 def GetSubDirectoryGroupedTaskGenerator(walker, subDirectoryGroupingDepth, directoryFiles, include, exclude, mapReduceFinalizeFunctions): 52 taskID = 0 53 for fileGroupSet in jobGenerator.GetAllMatchingFilesGroupedBySubdirectoryDepth(walker, directoryFiles, include, exclude, subDirectoryGroupingDepth): 54 task = RemoteProcessTask(taskID, walker) 55 taskID += 1 56 totalCount = 0 57 for filePath, relativePath, directoryPathStackForFile in fileGroupSet: 58 validMapReduceOperations = [ job.GetValidMapReduceOperations(relativePath) for job in mapReduceFinalizeFunctions ] 59 count = sum(map(len, validMapReduceOperations)) 60 if count: 61 log.debug('Adding %s to task', relativePath) 62 task.AddFileToTask(relativePath, directoryPathStackForFile, validMapReduceOperations) 63 totalCount += count 64 65 if totalCount: 66 log.info('Created directory grouped task with %i files', len(task.fileTasks)) 67 yield task 68 69 70 def ChunkIterable(iterable, partitionSize): 71 chunk = [] 72 for i in iterable: 73 chunk.append(i) 74 if len(chunk) == partitionSize: 75 yield chunk 76 chunk = [] 77 78 if len(chunk): 79 yield chunk 80 81 82 def MapReduceMany(walker, directoryFiles, include, exclude, mapReduceFinalizeFunctions, pool, subDirectoryGroupingDepth = None, filesPerBatchWhenNotGrouped = 20): 83 results = {} 84 outputFiles = {} 85 jobNames = set() 86 with benchmark('FSD Map Reduce'): 87 results = {} 88 for job in mapReduceFinalizeFunctions: 89 if job.name in jobNames: 90 raise ValueError('Job with name %s was duplicated in job list!' % job.name) 91 else: 92 jobNames.add(job.name) 93 if job.HasReduceFunction(): 94 results[job.name] = None 95 else: 96 results[job.name] = [] 97 98 if subDirectoryGroupingDepth is None: 99 taskIterator = GetSimpleTaskGenerator(walker, filesPerBatchWhenNotGrouped, directoryFiles, include, exclude, mapReduceFinalizeFunctions) 100 else: 101 taskIterator = GetSubDirectoryGroupedTaskGenerator(walker, subDirectoryGroupingDepth, directoryFiles, include, exclude, mapReduceFinalizeFunctions) 102 try: 103 with benchmark('FSD Map Dispatch'): 104 for r in pool.imap_unordered(PerformMapOperationsForAllJobsOnFileInOtherProcess, taskIterator, 1): 105 for i, job in enumerate(mapReduceFinalizeFunctions): 106 jobName = job.name 107 try: 108 if jobName in r: 109 if job.StoreOutputInTemporaryFiles(): 110 if job.name in outputFiles: 111 outputFiles[jobName].append(r[jobName]) 112 else: 113 outputFiles[jobName] = [r[jobName]] 114 else: 115 results[jobName] = job.Reduce(r[jobName], results[jobName]) 116 except: 117 log.critical('Error when reducing %s', jobName) 118 raise 119 120 except RemoteProcessException as e: 121 log.error('Exception occured on worker process:\n %s' % str(e)) 122 raise Exception('Terminating due to remote exception:\n' + str(e)) 123 124 for i, job in enumerate(mapReduceFinalizeFunctions): 125 jobName = job.name 126 with benchmark('Finalize %s' % jobName): 127 if job.StoreOutputInTemporaryFiles() and jobName in outputFiles: 128 results[jobName] = MergeOutputFromIntermediateFiles(pool, walker, outputFiles[jobName], job) 129 else: 130 results[jobName] = job.Finalize(walker.GetRoot(), results[jobName]) 131 132 return results 133 134 135 def MergeIntermediateFilesOnRemoteProcess(args): 136 walker, jobName, uniqueID, mergeFunction, reduceFunction, chunkedfileNames = args 137 outputFileName = '%s/Merge-%i.tmp' % (jobName, uniqueID) 138 outFile = walker.GetTemporaryFileForWriting(outputFileName, 'rb') 139 mergeFunction(outFile, reduceFunction, [ walker.OpenTemporaryFileForReading(f) for f in chunkedfileNames ]) 140 walker.CloseTemporaryWriteFiles() 141 return outputFileName 142 143 144 MAX_OPEN_FILES = 256 145 FILE_MERGE_CHUNK_SIZE = 10 146 147 def MergeOutputFromIntermediateFiles(pool, walker, mergeFiles, job): 148 intermediateFiles = mergeFiles 149 if len(intermediateFiles) < MAX_OPEN_FILES: 150 files = [ walker.OpenTemporaryFileForReading(f) for f in intermediateFiles ] 151 return job.ReduceAndFinalizeFromTemporaryFiles(files) 152 else: 153 mergeFunction = job.GetIncrementalFileMergeFunction() 154 uniqueCounter = itertools.count() 155 repeated_walker = itertools.repeat(walker) 156 repeated_job_name = itertools.repeat(job.name) 157 repeated_merge_fn = itertools.repeat(mergeFunction) 158 repeated_reduce_fn = itertools.repeat(job.reduceFunction) 159 while len(intermediateFiles) >= MAX_OPEN_FILES: 160 mergeBlocks = ChunkIterable(intermediateFiles, FILE_MERGE_CHUNK_SIZE) 161 newIntermediateFiles = [] 162 mapArguments = itertools.izip(repeated_walker, repeated_job_name, uniqueCounter, repeated_merge_fn, repeated_reduce_fn, mergeBlocks) 163 for resultFile in pool.imap_unordered(MergeIntermediateFilesOnRemoteProcess, mapArguments, 1): 164 newIntermediateFiles.append(resultFile) 165 166 intermediateFiles = newIntermediateFiles 167 168 files = [ walker.OpenTemporaryFileForReading(f) for f in intermediateFiles ] 169 return job.ReduceAndFinalizeFromTemporaryFiles(files) 170 171 172 class RemoteProcessTask(object): 173 174 def __init__(self, uniqueTaskID, walker): 175 self.uniqueTaskID = uniqueTaskID 176 self.walker = walker 177 self.fileTasks = [] 178 179 def AddFileToTask(self, relativePath, directoryFilePaths, jobs): 180 self.fileTasks.append([relativePath, directoryFilePaths, jobs]) 181 182 def Execute(self): 183 resultsByJob = {} 184 listMergeFunctionsByJob = {} 185 reduceByJob = {} 186 directoryFiles = {} 187 persistFunctionsByJob = {} 188 for relativePath, directoryFilePaths, jobs in self.fileTasks: 189 for f in directoryFilePaths: 190 if f not in directoryFiles: 191 log.info('Opening directory file %s', f) 192 contents = self.walker.GetFileContents(AbsJoin(self.walker.GetRoot(), f)) 193 directoryFiles[f] = contents 194 195 if len(directoryFilePaths) > 0: 196 curDir = directoryFilePaths[-1] 197 if not self.walker.IsFolderIncluded(directoryFiles[curDir]): 198 continue 199 path = AbsJoin(self.walker.GetRoot(), relativePath) 200 with benchmark('File Open and Read: %s' % path): 201 data = self.walker.GetFileContents(path) 202 for jobIndex, job in enumerate(jobs): 203 for jobName, mapName, mapFunction, mapKwargs, reduceFunction, isGeneratorFunction, listMergeFunction, persistFunction in job: 204 if reduceFunction is not None: 205 reduceByJob[jobName] = reduceFunction 206 if jobName not in resultsByJob: 207 resultsByJob[jobName] = [] 208 if persistFunction is not None: 209 persistFunctionsByJob[jobName] = persistFunction 210 if listMergeFunction is not None: 211 listMergeFunctionsByJob[jobName] = listMergeFunction 212 stack = [ directoryFiles[i] for i in directoryFilePaths ] 213 with benchmark('Map: %s' % mapName): 214 try: 215 if mapKwargs is None: 216 mapKwargs = {} 217 functionResult = mapFunction(self.walker.GetRoot(), relativePath, stack, data, **mapKwargs) 218 except Exception as e: 219 raise RemoteProcessException('Exception while running map function "%s" on file: %s \n%s' % (mapName, path, traceback.format_exc())) 220 221 if isGeneratorFunction: 222 try: 223 allResults = [ r for r in functionResult ] 224 except Exception as e: 225 raise RemoteProcessException('Exception while running map function "%s" on file: %s \n%s' % (mapName, path, traceback.format_exc())) 226 227 resultsByJob[jobName].extend(allResults) 228 else: 229 resultsByJob[jobName].append(functionResult) 230 231 for jobName in resultsByJob.keys(): 232 if jobName in persistFunctionsByJob: 233 temporaryFileName = '%s/%i.tmp' % (jobName, self.uniqueTaskID) 234 log.info('Writing results to intermediate file: %s', temporaryFileName) 235 f = self.walker.GetTemporaryFileForWriting(temporaryFileName, 'wb') 236 listReduceFunction = listMergeFunctionsByJob[jobName] 237 if jobName in reduceByJob: 238 iterableResults = listReduceFunction([ i[1] for i in resultsByJob[jobName] ]) 239 reducibleResults = [ i[0] for i in resultsByJob[jobName] ] 240 reduceFunction = reduceByJob[jobName] 241 reducedResults = reduceFunction(None, reducibleResults) 242 persistFunctionsByJob[jobName](f, reducedResults, len(iterableResults), iterableResults) 243 else: 244 iterableResults = listReduceFunction(resultsByJob[jobName]) 245 persistFunctionsByJob[jobName](f, None, len(iterableResults), iterableResults) 246 resultsByJob[jobName] = temporaryFileName 247 elif jobName in reduceByJob: 248 resultsByJob[jobName] = [reduceByJob[jobName](None, resultsByJob[jobName])] 249 250 if len(persistFunctionsByJob): 251 self.walker.CloseTemporaryWriteFiles() 252 return resultsByJob 253 254 255 def PerformMapOperationsForAllJobsOnFileInOtherProcess(a): 256 return a.Execute()