title: WriteToBigQuery with dynamic destinations
body: <p>I am working on an Apache Beam pipeline that reads a bunch of events from a pub/sub and then based on event type, it writes them into separate BigQuery tables.</p>
  <p>I know that <code>WriteToBigQuery</code> supports dynamic destinations but the problem in my case is that the destination is derived from the data that is read from the event. For example:
  an event looks like</p>
  <pre><code>{
   &quot;object_id&quot;: 123,
   ... some metadata,
   &quot;object_data&quot;: {object related info}
  }
  </code></pre>
  <p>Data that should be written to BigQuery table is under the <code>object_data</code> key from the event, but, the table name is derived from other fields in the metadata.
  I tried to use the side inputs params but the issue is that because each event can have different destinations, the side inputs don't update accordingly. The code is below:</p>
  <pre><code>class DumpToBigQuery(PTransform):
  
      def _choose_table(self, element, table_names):
          # table_names = {&quot;table_name&quot;: &quot;project_name.dataset.table_name}
          table_name = table_names[&quot;table_name&quot;]
          return table_name
  
      def expand(self, pcoll):
          events = (
              pcoll
              | &quot;GroupByObjectType&quot; &gt;&gt; Map(lambda e: (e[&quot;object_type&quot;], e))
              | &quot;Window&quot;
              &gt;&gt; WindowInto(
                  windowfn=FixedWindows(self.window_interval_seconds)
              )
              | &quot;GroupByKey&quot; &gt;&gt; GroupByKey()
              | &quot;KeepLastEventOnly&quot; &gt;&gt; ParDo(WillTakeLatestEventForKey()
          )
  
          table_name = events | Map(lambda e: [&quot;table_name&quot;, f&quot;{self.project}:{self.dataset}.{e[0]}&quot;])
          table_names_dct = AsDict(table_name)
  
          events_to_write = events | Map(lambda e: e[1]) | Map(self._drop_unwanted_fields)
  
          return events_to_write | &quot;toBQ&quot; &gt;&gt; WriteToBigQuery(
              table=self._choose_table,
              table_side_inputs=(table_names_dct,),
              create_disposition=BigQueryDisposition.CREATE_NEVER,
              insert_retry_strategy=RetryStrategy.RETRY_NEVER,
          )
  
  </code></pre>
  <p>You can see that the side input is taken from the other branch of the pipeline <code>table_name</code> that is basically extracting the table name from the event. And then, this is given as input to <code>WriteToBigQuery</code>. Unfortunately, this doesn't really work as under load, the side input is not updated and some events are using wrong destinations.</p>
  <p>What other approach I can use in this specific case? All the docs are using static examples and don't really cover this dynamic approach.</p>
  <p>The other thing I tried is to write a custom <code>DoFn</code> that used the HTTP BigQuery client and inserts the rows, the issue here is the speed of the pipeline as is inserting around 6-7 events per seconds.</p>