parameters: (self, ranges, mapper, reducer)
return_statement: return final_results.compute()
docstring: Performs map-reduce using Dask framework.
  
          Args:
              mapper (function): A function that runs the computational graph
                  and returns a list of values.
  
              reducer (function): A function that merges two lists that were
                  returned by the mapper.
  
          Returns:
              list: A list representing the values of action nodes returned
              after computation (Map-Reduce).
docstring_summary: Performs map-reduce using Dask framework.
docstring_tokens: Performs map - reduce using Dask framework .
function: def ProcessAndMerge(self, ranges, mapper, reducer):
          """
          Performs map-reduce using Dask framework.
  
          Args:
              mapper (function): A function that runs the computational graph
                  and returns a list of values.
  
              reducer (function): A function that merges two lists that were
                  returned by the mapper.
  
          Returns:
              list: A list representing the values of action nodes returned
              after computation (Map-Reduce).
          """
  
          # These need to be passed as variables, because passing `self` inside
          # following `dask_mapper` function would trigger serialization errors
          # like the following:
          #
          # AttributeError: Can't pickle local object 'DaskBackend.ProcessAndMerge.<locals>.dask_mapper'
          # TypeError: cannot pickle '_asyncio.Task' object
          #
          # Which boil down to the self.client object not being serializable
          headers = self.headers
          shared_libraries = self.shared_libraries
  
          def dask_mapper(current_range):
              """
              Gets the paths to the file(s) in the current executor, then
              declares the headers found.
  
              Args:
                  current_range (tuple): The current range of the dataset being
                      processed on the executor.
  
              Returns:
                  function: The map function to be executed on each executor,
                  complete with all headers needed for the analysis.
              """
              # Retrieve the current worker local directory
              localdir = get_worker().local_directory
  
              # Get and declare headers on each worker
              headers_on_executor = [
                  os.path.join(localdir, os.path.basename(filepath))
                  for filepath in headers
              ]
              Utils.declare_headers(headers_on_executor)
  
              # Get and declare shared libraries on each worker
              shared_libs_on_ex = [
                  os.path.join(localdir, os.path.basename(filepath))
                  for filepath in shared_libraries
              ]
              Utils.declare_shared_libraries(shared_libs_on_ex)
  
              return mapper(current_range)
  
          dmapper = dask.delayed(dask_mapper)
          dreducer = dask.delayed(reducer)
  
          mergeables_lists = [dmapper(range) for range in ranges]
  
          while len(mergeables_lists) > 1:
              mergeables_lists.append(
                  dreducer(mergeables_lists.pop(0), mergeables_lists.pop(0)))
  
          # Here we start the progressbar for the current RDF computation graph
          # running on the Dask client. This expects a future object, so we need
          # convert the last delayed object from the list above to a future
          # through the `persist` call. This also starts the computation in the
          # background, but the time difference is negligible. The progressbar is
          # properly shown in the terminal, whereas in the notebook it can be
          # shown only if it's the last call in a cell. Since we're encapsulating
          # it in this class, it won't be shown. Full details at
          # https://docs.dask.org/en/latest/diagnostics-distributed.html#dask.distributed.progress
          final_results = mergeables_lists.pop().persist()
          progress(final_results)
  
          return final_results.compute()
function_tokens: def ProcessAndMerge ( self , ranges , mapper , reducer ) : # These need to be passed as variables, because passing `self` inside # following `dask_mapper` function would trigger serialization errors # like the following: # # AttributeError: Can't pickle local object 'DaskBackend.ProcessAndMerge.<locals>.dask_mapper' # TypeError: cannot pickle '_asyncio.Task' object # # Which boil down to the self.client object not being serializable headers = self . headers shared_libraries = self . shared_libraries def dask_mapper ( current_range ) : """
              Gets the paths to the file(s) in the current executor, then
              declares the headers found.
  
              Args:
                  current_range (tuple): The current range of the dataset being
                      processed on the executor.
  
              Returns:
                  function: The map function to be executed on each executor,
                  complete with all headers needed for the analysis.
              """ # Retrieve the current worker local directory localdir = get_worker ( ) . local_directory # Get and declare headers on each worker headers_on_executor = [ os . path . join ( localdir , os . path . basename ( filepath ) ) for filepath in headers ] Utils . declare_headers ( headers_on_executor ) # Get and declare shared libraries on each worker shared_libs_on_ex = [ os . path . join ( localdir , os . path . basename ( filepath ) ) for filepath in shared_libraries ] Utils . declare_shared_libraries ( shared_libs_on_ex ) return mapper ( current_range ) dmapper = dask . delayed ( dask_mapper ) dreducer = dask . delayed ( reducer ) mergeables_lists = [ dmapper ( range ) for range in ranges ] while len ( mergeables_lists ) > 1 : mergeables_lists . append ( dreducer ( mergeables_lists . pop ( 0 ) , mergeables_lists . pop ( 0 ) ) ) # Here we start the progressbar for the current RDF computation graph # running on the Dask client. This expects a future object, so we need # convert the last delayed object from the list above to a future # through the `persist` call. This also starts the computation in the # background, but the time difference is negligible. The progressbar is # properly shown in the terminal, whereas in the notebook it can be # shown only if it's the last call in a cell. Since we're encapsulating # it in this class, it won't be shown. Full details at # https://docs.dask.org/en/latest/diagnostics-distributed.html#dask.distributed.progress final_results = mergeables_lists . pop ( ) . persist ( ) progress ( final_results ) return final_results . compute ( )