title: How can I merge two Pcollections(various sizes/data) with a common "key"(Street) via a side input?
body: <p>I have two PCollections: one that pulls info from Pub/Sub, and one that pulls data from a CSV file. After some various transforms in each pipeline, I'd like to merge the two on a common key they both share, "STREET". I'm including the second PCollection as a side input. However, I'm getting an error on attempt to run.</p>
  
  <p>I've attempted to utilize CoGroupByKey, but I kept receiving errors regarding the difference in types of data in the Pcollections. I tried refactoring the outputs, and setting the attribute of the PCollection via the <code>__setattr__</code> to force the types to be equal, but it reported "mixed values" regardless. After further research, it seems like it's better to utilize side inputs, especially when there's disparity in data size between elements. Even with side inputs, I'm still unable to get past the current error: </p>
  
  <pre><code>from_runner_api raise ValueError('No producer for %s' % id)
  ValueError: No producer for ref_PCollection_PCollection_6
  </code></pre>
  
  <p>My application logic is as follows:</p>
  
  <pre><code>def merge_accidents(element, pcoll):
      print(element)
      print(pcoll)
      "some code that will append to existing data"
  
  accident_pl = beam.Pipeline()
  accident_data = (accident_pl |
                          'Read' &gt;&gt; beam.io.ReadFromText('/modified_Excel_Crashes_Chicago.csv')
                          | 'Map Accidents' &gt;&gt; beam.ParDo(AccidentstoDict())
                          | 'Count Accidents' &gt;&gt; Count.PerKey())
  
  chi_traf_pl = beam.Pipeline(options=pipeline_options)
  chi_traffic = (chi_traf_pl | 'ReadPubSub' &gt;&gt; beam.io.ReadFromPubSub(subscription=subscription_name, with_attributes=True)
                             | 'GeoEnrich&amp;Trim' &gt;&gt; beam.Map(loc_trim_enhance)
                             | 'TimeDelayEnrich' &gt;&gt; beam.Map(timedelay)
                             | 'TrafficRatingEnrich' &gt;&gt; beam.Map(traffic_rating)
                             | 'MergeAccidents' &gt;&gt; beam.Map(merge_accidents, pcoll=AsDict(accident_data))
                             | 'Temp Write'&gt;&gt; beam.io.WriteToText('testtime', file_name_suffix='.txt'))
  
  accident_pl.run()
  chi_result = chi_traf_pl.run()
  chi_result.wait_until_finish()```
  
  **Pcoll 1:**
  [{'segment_id': '1', 'street': 'Western Ave', 'direction': 'EB', 'length': '0.5', 'cur_traffic': '24', 'county': 'Cook County', 'neighborhood': 'West Elsdon', 'zip_code': '60629', 'evnt_timestamp': '2019-04-01 20:50:20.0', 'traffic_rating': 'Heavy', 'time_delay': '0.15'}]
  **Pcoll 2:**
  ('MILWAUKEE AVE', 1)
  ('CENTRAL AVE', 2)
  ('WESTERN AVE', 6)
  
  **Expected:**
  [{'segment_id': '1', 'street': 'Western Ave', 'direction': 'EB', 'length': '0.5', 'cur_traffic': '24', 'county': 'Cook County', 'neighborhood': 'West Elsdon', 'zip_code': '60629', 'evnt_timestamp': '2019-04-01 20:50:20.0', 'traffic_rating': 'Heavy', 'time_delay': '0.15', 'accident_count': '6'}]
  
  **Actual Results:**
  "from_runner_api raise ValueError('No producer for %s' % id)ValueError: No producer for ref_PCollection_PCollection_6
  </code></pre>