title: Calling beam.io.WriteToBigQuery in a beam.DoFn
body: <p>I've created a dataflow template with some parameters. When I write the data to BigQuery, I would like to make use of these parameters to determine which table it is supposed to write to. I've tried calling WriteToBigQuery in a ParDo as  suggested in the following link. </p>
  
  <p><a href="https://stackoverflow.com/questions/59217902/how-can-i-write-to-big-query-using-a-runtime-value-provider-in-apache-beam">How can I write to Big Query using a runtime value provider in Apache Beam?</a></p>
  
  <p>The pipeline ran successfully but it is not creating or loading data to BigQuery. Any idea what might be the issue?</p>
  
  <pre><code>def run():
    pipeline_options = PipelineOptions()
    pipeline_options.view_as(DebugOptions).experiments = ['use_beam_bq_sink']
  
    with beam.Pipeline(options=pipeline_options) as p:
      custom_options = pipeline_options.view_as(CustomOptions)
  
      _ = (
        p
        | beam.Create([None])
        | 'Year to periods' &gt;&gt; beam.ParDo(SplitYearToPeriod(custom_options.year))
        | 'Read plan data' &gt;&gt; beam.ParDo(GetPlanDataByPeriod(custom_options.secret_name))
        | 'Transform record' &gt;&gt; beam.Map(transform_record)
        | 'Write to BQ' &gt;&gt; beam.ParDo(WritePlanDataToBigQuery(custom_options.year))
      )
  
  if __name__ == '__main__':
    run()
  </code></pre>
  
  <pre><code>class CustomOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
      parser.add_value_provider_argument('--year', type=int)
      parser.add_value_provider_argument('--secret_name', type=str)
  </code></pre>
  
  <pre><code>class WritePlanDataToBigQuery(beam.DoFn):
    def __init__(self, year_vp):
      self._year_vp = year_vp
  
    def process(self, element):
      year = self._year_vp.get()
  
      table = f's4c.plan_data_{year}'
      schema = {
        'fields': [ ...some fields properties ]
      }
  
      beam.io.WriteToBigQuery(
        table=table,
        schema=schema,
        create_disposition=BigQueryDisposition.CREATE_IF_NEEDED,
        write_disposition=BigQueryDisposition.WRITE_TRUNCATE,
        method=beam.io.WriteToBigQuery.Method.FILE_LOADS
      )
  </code></pre>