docs: def Flush(self, state):
      """Finish writing JSON files, upload to cloudstorage and bigquery."""
      self.bigquery = bigquery.GetBigQueryClient()
      # BigQuery job ids must be alphanum plus dash and underscore.
      urn_str = self.source_urn.RelativeName("aff4:/").replace("/", "_").replace(
          ":", "").replace(".", "-")
  
      for tracker in itervalues(self.temp_output_trackers):
        # Close out the gzip handle and pass the original file handle to the
        # bigquery client so it sees the gzip'd content.
        tracker.gzip_filehandle.write("\n")
        tracker.gzip_filehandle.close()
        tracker.gzip_filehandle_parent.seek(0)
  
        # e.g. job_id: hunts_HFFE1D044_Results_ExportedFile_1446056474
        job_id = "{0}_{1}_{2}".format(
            urn_str, tracker.output_type,
            rdfvalue.RDFDatetime.Now().AsSecondsSinceEpoch())
  
        # If we have a job id stored, that means we failed last time. Re-use the
        # job id and append to the same file if it continues to fail. This avoids
        # writing many files on failure.
        if tracker.output_type in self.output_jobids:
          job_id = self.output_jobids[tracker.output_type]
        else:
          self.output_jobids[tracker.output_type] = job_id
  
        if (state.failure_count + self.failure_count >=
            config.CONFIG["BigQuery.max_upload_failures"]):
          logging.error(
              "Exceeded BigQuery.max_upload_failures for %s, giving up.",
              self.source_urn)
        else:
          try:
            self.bigquery.InsertData(tracker.output_type,
                                     tracker.gzip_filehandle_parent,
                                     tracker.schema, job_id)
            self.failure_count = max(0, self.failure_count - 1)
            del self.output_jobids[tracker.output_type]
          except bigquery.BigQueryJobUploadError:
            self.failure_count += 1
  
      # Now that everything is in bigquery we can remove the output streams
      self.temp_output_trackers = {}