title: Partitioned table loading using DataFlow job
body: <p>I want to read the file and need to write it to BigQuery Partitioned table, based on the date value present in the field of the file. e.g. If file contains 2 dates 25 and 26 July then DataFlow should write that data to 2 partitions based on the data present in the file.</p>
  
  
  
  <pre class="lang-java prettyprint-override"><code>public class StarterPipeline {
    private static final Logger LOG =
        LoggerFactory.getLogger(StarterPipeline.class);
  
    public static void main(String[] args) {
      DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
      options.setProject("");
      options.setTempLocation("gs://stage_location/");
      Pipeline p = Pipeline.create(options);
  
      List&lt;TableFieldSchema&gt; fields = new ArrayList&lt;&gt;();
      fields.add(new TableFieldSchema().setName("id").setType("STRING"));
      fields.add(new TableFieldSchema().setName("name").setType("STRING"));
      fields.add(new TableFieldSchema().setName("designation").setType("STRING"));
      fields.add(new TableFieldSchema().setName("joindate").setType("STRING"));
      TableSchema schema = new TableSchema().setFields(fields);
  
      PCollection&lt;String&gt; read = p.apply("Read Lines",TextIO.read().from("gs://hadoop_source_files/employee.txt"));
  
      PCollection&lt;TableRow&gt; rows = read.apply(ParDo.of(new DoFn&lt;String,TableRow&gt;(){
        @ProcessElement
        public void processElement(ProcessContext c) {
          String[] data = c.element().split(",");
  
          c.output(new TableRow().set("id", data[0]).set("name", data[1]).set("designation", data[2]).set("joindate", data[3]));
        }
      }));
  
  
      rows.apply(BigQueryIO.writeTableRows().to(new SerializableFunction&lt;ValueInSingleWindow&lt;TableRow&gt;, TableDestination&gt;() {
        public String getDate(String value) {
          return "project:dataset.DataFlow_Test$"+value;
        }
  
        @Override
        public TableDestination apply(ValueInSingleWindow&lt;TableRow&gt; value) {
          TableRow row = value.getValue();
          String date = getDate(row.get("joindate").toString());
          String tableSpec = date;
          String tableDescription = "";
          return new TableDestination(tableSpec, tableDescription);
        }
      }).withFormatFunction(new SerializableFunction&lt;TableRow, TableRow&gt;() {
        @Override
        public TableRow apply(TableRow input) {
          // TODO Auto-generated method stub
          return input;
        }
      }).withSchema(schema)
          .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
          .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));
  
      p.run();
    }
  }
  </code></pre>
  
  <p>while running above program I am getting below error: Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.IllegalArgumentException: Table reference is not in [project_id]:[dataset_id].[table_id] format: Caused by: java.lang.IllegalArgumentException: Table reference is not in [project_id]:[dataset_id].[table_id] format. Let me know if there are any recommendations</p>