parameters: (self, ranges, mapper, reducer)
return_statement: return parallel_collection.map(spark_mapper).treeReduce(reducer)
docstring: Performs map-reduce using Spark 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 Spark framework.
docstring_tokens: Performs map - reduce using Spark framework .
function: def ProcessAndMerge(self, ranges, mapper, reducer):
          """
          Performs map-reduce using Spark 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 and not as class attributes
          # otherwise the `spark_mapper` function would be referencing this
          # this instance of the Spark backend along with the referenced
          # SparkContext. This would cause the errors described in SPARK-5063.
          headers = self.headers
          shared_libraries = self.shared_libraries
  
          def spark_mapper(current_range):
              """
              Gets the paths to the file(s) in the current executor, then
              declares the headers found.
  
              Args:
                  current_range (tuple): A pair that contains the starting and
                      ending values of the current range.
  
              Returns:
                  function: The map function to be executed on each executor,
                  complete with all headers needed for the analysis.
              """
              # Get and declare headers on each worker
              headers_on_executor = [
                  pyspark.SparkFiles.get(ntpath.basename(filepath))
                  for filepath in headers
              ]
              Utils.declare_headers(headers_on_executor)
  
              # Get and declare shared libraries on each worker
              shared_libs_on_ex = [
                  pyspark.SparkFiles.get(ntpath.basename(filepath))
                  for filepath in shared_libraries
              ]
              Utils.declare_shared_libraries(shared_libs_on_ex)
  
              return mapper(current_range)
  
          # Build parallel collection
          parallel_collection = self.sc.parallelize(ranges, len(ranges))
  
          # Map-Reduce using Spark
          return parallel_collection.map(spark_mapper).treeReduce(reducer)
function_tokens: def ProcessAndMerge ( self , ranges , mapper , reducer ) : # These need to be passed as variables and not as class attributes # otherwise the `spark_mapper` function would be referencing this # this instance of the Spark backend along with the referenced # SparkContext. This would cause the errors described in SPARK-5063. headers = self . headers shared_libraries = self . shared_libraries def spark_mapper ( current_range ) : """
              Gets the paths to the file(s) in the current executor, then
              declares the headers found.
  
              Args:
                  current_range (tuple): A pair that contains the starting and
                      ending values of the current range.
  
              Returns:
                  function: The map function to be executed on each executor,
                  complete with all headers needed for the analysis.
              """ # Get and declare headers on each worker headers_on_executor = [ pyspark . SparkFiles . get ( ntpath . basename ( filepath ) ) for filepath in headers ] Utils . declare_headers ( headers_on_executor ) # Get and declare shared libraries on each worker shared_libs_on_ex = [ pyspark . SparkFiles . get ( ntpath . basename ( filepath ) ) for filepath in shared_libraries ] Utils . declare_shared_libraries ( shared_libs_on_ex ) return mapper ( current_range ) # Build parallel collection parallel_collection = self . sc . parallelize ( ranges , len ( ranges ) ) # Map-Reduce using Spark return parallel_collection . map ( spark_mapper ) . treeReduce ( reducer )