title: Error while splitting pcollections on Dataflow runner
body: <p>I have an Apache Beam pipeline built in python. I am reading rows from a csv file.Then there are generic pipeline steps for All pcollections. This works fine. For pcollections which come from a specific filename, I want to perform couple of additional steps. 
  Therefore I tag the pcollections in that file and run additional steps for those tagged collections.
  Wehn I run the pipeline on 'Dataflow' it gives me the error "Workflow failed. Causes: Expected custom source to have non-zero number of splits. "</p>
  
  <p>I tested and this works fine on 'DirectRunner'. </p>
  
  <pre><code>lines = (p | beam.io.ReadFromText(input_file_path, skip_header_lines=1))
  
  Generic = (lines | &lt;"Do generic logic for all pCollections"&gt;)
  
  tagged_lines = (lines | beam.ParDo(Tag(),input_file_path).with_outputs(Tag.TAG_OPTOUT,Tag.TAG_BOUNCE))
  
  Optouts = (tagged_lines[Tag.TAG_OPTOUT] | &lt;"Do logic 1"&gt;)
  
  Bounces = (tagged_lines[Tag.TAG_BOUNCE] | &lt;"Do logic 2"&gt;)
  
  </code></pre>
  
  <pre><code>class Tag(beam.DoFn):
      TAG_OPTOUT = 'OPTOUT'
      TAG_BOUNCE = 'BOUNCE'
      def process(self, element,input_file_path):
          input_file = input_file_path.get()
          if "optout" in input_file:
              yield pvalue.TaggedOutput(self.TAG_OPTOUT, element)
          elif "bounce" in input_file:
          yield pvalue.TaggedOutput(self.TAG_BOUNCE, element)
  </code></pre>