title: Google Dataflow / Apache Beam Python - Side-Input from PCollection kills performance
body: <p>We are running logfile parsing jobs in google dataflow using the Python SDK. Data is spread over several 100s of daily logs, which we read via file-pattern from Cloud Storage. Data volume for all files is about 5-8 GB (gz files) with 50-80 million lines in total.</p>
  
  <pre><code>loglines = p | ReadFromText('gs://logfile-location/logs*-20180101')
  </code></pre>
  
  <p>In addition, we have a simple (small) mapping csv, that maps logfile-entries to human readable text. Has about 400 lines, 5 kb size. </p>
  
  <p>For Example a logfile entry with [param=testing2] should be mapped to "Customer requested 14day free product trial" in the final output.</p>
  
  <p>We do this in a simple beam.Map with sideinput, like so:</p>
  
  <pre><code>customerActions = loglines | beam.Map(map_logentries,mappingTable)
  </code></pre>
  
  <p>where map_logentries is the mapping function and mappingTable is said mapping table.</p>
  
  <p>However, this only works if we read the mapping table in native python via open() / read(). If we do the same utilising the beam pipeline via ReadFromText() and pass the resulting PCollection as side-input to the Map, like so:</p>
  
  <pre><code>mappingTable = p | ReadFromText('gs://side-inputs/category-mapping.csv')    
  customerActions = loglines | beam.Map(map_logentries,beam.pvalue.AsIter(mappingTable))
  </code></pre>
  
  <p>performance breaks down completely to about 2-3 items per Second.</p>
  
  <p>Now, my questions:</p>
  
  <ol>
  <li>Why would performance break so badly, what is wrong with passing a
  PCollection as side-input? </li>
  <li>If it is maybe not recommended to use
  PCollections as side-input, how is one supposed to build such as
  pipeline that needs mappings that can/should not be hard coded into
  the mapping function?</li>
  </ol>
  
  <p>For us, the mapping does change frequently and I need to find a way to have "normal" users provide it. The idea was to have the mapping csv available in Cloud Storage, and simply incorporate it into the Pipeline via ReadFromText(). Reading it locally involves providing the mapping to the workers, so only the tech-team can do this.</p>
  
  <p>I am aware that there are caching issues with side-input, but surely this should not apply to a 5kb input.</p>
  
  <p>All code above is pseudo code to explain the problem. Any ideas and thoughts on this would be highly appreciated!</p>