NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

operations.py (5003B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\fsd\mapReduce\operations.py
      2 from __future__ import absolute_import
      3 import logging
      4 import inspect
      5 import heapq
      6 from .iterativePickle import IterativePickleWithHeaderObject
      7 import cPickle
      8 log = logging.getLogger(__name__)
      9 log.addHandler(logging.NullHandler())
     10 
     11 class MapOperation(object):
     12 
     13     def __init__(self, name, mapFunction, pathConditional = None, mapKwargs = None):
     14         self.name = name
     15         self.conditional = pathConditional
     16         self.mapFunction = mapFunction
     17         self.isGenerator = inspect.isgeneratorfunction(mapFunction)
     18         self.mapKwargs = mapKwargs
     19         if mapKwargs:
     20             cPickle.dumps(mapKwargs)
     21 
     22     def IsValidOnPath(self, path):
     23         if self.conditional is None:
     24             return True
     25         elif self.conditional.Matches(path):
     26             log.debug('Running Map %s on %s (matches conditional)', self.name, path)
     27             return True
     28         else:
     29             log.debug('Skipping Map %s on %s due to path conditional', self.name, path)
     30             return False
     31 
     32 
     33 class MapReduceFinalizeJob(object):
     34 
     35     def __init__(self, name, mapOperations, pathConditional = None, reduceFunction = None, finalizeFunction = None):
     36         self.name = name
     37         self.conditional = pathConditional
     38         self.mapOperations = mapOperations
     39         self.reduceFunction = reduceFunction
     40         self.finalizeFunction = finalizeFunction
     41 
     42     def GetValidMapReduceOperations(self, path):
     43         if self.conditional is not None and not self.conditional.Matches(path):
     44             log.debug('Skipping job %s on %s due to job path conditional', self.name, path)
     45             return []
     46         validMapOperations = []
     47         for mapOperation in self.mapOperations:
     48             if mapOperation.IsValidOnPath(path):
     49                 log.debug('Map operation %s pending on %s', mapOperation.name, path)
     50                 validMapOperations.append(self.GetOperationInfo(mapOperation))
     51 
     52         return validMapOperations
     53 
     54     def GetOperationInfo(self, mapOperation):
     55         return (self.name,
     56          mapOperation.name,
     57          mapOperation.mapFunction,
     58          mapOperation.mapKwargs,
     59          self.reduceFunction,
     60          mapOperation.isGenerator,
     61          None,
     62          None)
     63 
     64     def HasReduceFunction(self):
     65         return self.reduceFunction is not None
     66 
     67     def StoreOutputInTemporaryFiles(self):
     68         return False
     69 
     70     def Reduce(self, data, existingResult):
     71         if self.reduceFunction is not None:
     72             return self.reduceFunction(existingResult, data)
     73         else:
     74             existingResult.extend(data)
     75             return existingResult
     76 
     77     def Finalize(self, rootPath, data):
     78         if self.finalizeFunction is None:
     79             return data
     80         else:
     81             return self.finalizeFunction(rootPath, data)
     82 
     83 
     84 def SaveResultsToFileObject(fileobject, reducedSummary, count, sortedResults):
     85     IterativePickleWithHeaderObject.WriteIterableWithHeaderAndLengthAsIterativePickle(fileobject, reducedSummary, count, sortedResults)
     86 
     87 
     88 def MergeSortedIntermediateFiles(outfile, reduceFunction, mergeFiles):
     89     files = [ IterativePickleWithHeaderObject(i) for i in mergeFiles ]
     90     totalCount = sum(map(len, files))
     91     if reduceFunction is not None:
     92         summariesForReducing = [ i.header for i in files ]
     93         reducedSummary = reduceFunction(None, summariesForReducing)
     94     else:
     95         reducedSummary = None
     96     IterativePickleWithHeaderObject.WriteIterableWithHeaderAndLengthAsIterativePickle(outfile, reducedSummary, totalCount, heapq.merge(*files))
     97 
     98 
     99 class MapReduceSortedStreamJob(MapReduceFinalizeJob):
    100 
    101     def __init__(self, name, mapOperations, incrementalFinalizeFunction, pathConditional = None, reduceFunction = None):
    102         MapReduceFinalizeJob.__init__(self, name, mapOperations, pathConditional, reduceFunction=reduceFunction, finalizeFunction=None)
    103         self.incrementalFinalizeFunction = incrementalFinalizeFunction
    104 
    105     def GetOperationInfo(self, mapOperation):
    106         return (self.name,
    107          mapOperation.name,
    108          mapOperation.mapFunction,
    109          mapOperation.mapKwargs,
    110          self.reduceFunction,
    111          mapOperation.isGenerator,
    112          sorted,
    113          SaveResultsToFileObject)
    114 
    115     def StoreOutputInTemporaryFiles(self):
    116         return True
    117 
    118     def GetIncrementalFileMergeFunction(self):
    119         return MergeSortedIntermediateFiles
    120 
    121     def ReduceAndFinalizeFromTemporaryFiles(self, fileObjects):
    122         files = [ IterativePickleWithHeaderObject(i) for i in fileObjects ]
    123         totalCount = sum(map(len, files))
    124         if self.HasReduceFunction():
    125             summariesForReducing = [ i.header for i in files ]
    126             reducedSummary = self.reduceFunction(None, summariesForReducing)
    127             return self.incrementalFinalizeFunction(totalCount, reducedSummary, heapq.merge(*files))
    128         else:
    129             return self.incrementalFinalizeFunction(totalCount, None, heapq.merge(*files))