title: Part of MapReduce lib (PipelineBase) sporadically unavailable in task queue, eventually becomes unavailable globally
body: <p><strong>Problem:</strong>
  We are seeing the following problem after upgrading the MapReduce library to the latest version (we had a version using the File API and we switched when that became deprecated):</p>
  
  <p>When running the MapReducer, in the tasks it’s starting, PipelineBase from the library sporadically becomes unavailable, forcing the task to fail. At the retry, it works ok. After a random period of time has passed, said part of the MapReduce library becomes completely and globally unavailable. This means it’s not available in the normal application background, nor in the tasks.</p>
  
  <p>The only way to make the application respond again is to reset all instances and make them reload the whole application code from scratch.</p>
  
  <p><strong>Error:</strong></p>
  
  <pre><code>File "/process/custom_mapper/views.py", line 23, in &lt;module&gt; from process.custom_mapper.pipeline import CustomMapperPipeline 
  
  File "/process/custom_mapper/pipeline.py", line 7, in &lt;module&gt; class CustomMapperPipeline(PipelineBase): 
  TypeError: Error when calling the metaclass bases cannot create 'NoneType’ instances
  </code></pre>
  
  <p>Which implies that PipelineBase is None. Looking inside the MapReduce library, in base_handler.py, from where PipelineBase is imported, we can see this code:</p>
  
  <pre><code>try:
    from mapreduce import pipeline_base
  except ImportError:
    pipeline_base = None
  </code></pre>
  
  <p>And then:</p>
  
  <pre><code>if pipeline_base:
    # For backward compatiblity.
    PipelineBase = pipeline_base.PipelineBase
  else:
    PipelineBase = None
  </code></pre>
  
  <p>Our assumption is that the import from mapreduce import pipeline_base is failing.</p>
  
  <p><strong>Extra info:</strong>
  The app is using Flask and the MapReducer is started remotely via an external endpoint. This endpoint is defined inside urls.py, loading a mapper controller class. This class imports the CustomMapperPipeline module. This means that each time an url is accessed, the CustomMapperPipeline class is instantiated. If PipelineBase is None, then the whole app fails.</p>
  
  <p>At some point in time PipelineBase becomes completely unavailable.</p>
  
  <p>As per the instructions, the mapreducer package is in the root of the app, next to the app.yaml file.</p>
  
  <p>There are only 2 items sent to the MapReducer currently.</p>
  
  <p><strong>Our implementation:</strong>
  The map reducer goes through a list of items. Each item will create a set of X number task queues to process data. Each of the X task queues will create Y number of task queues themselves to process data.</p>
  
  <p>We are using the following trigger for the map reduce:</p>
  
  <pre><code>yield MapreducePipeline(
      job_name,
      'process.custom_mapper.mappers.map_items',
      'process.custom_mapper.mappers.reduce_items',
      'process.custom_mapper.readers.ItemInputReader',
      'mapreduce.output_writers.GoogleCloudStorageOutputWriter',
      mapper_params={
          [...]
      },
      reducer_params={
          'output_writer': {
              'mime_type': 'text/plain',
              'bucket_name': bucket_name
          }
      }
  )
  </code></pre>
  
  <p>The CustomInputReader makes sure that each item is distributed to an individual shard.</p>