@Test public void testExportBigQueryToGcsSingleShardAwaitCompletion() throws Exception { when(mockBigqueryTablesGet.execute()).thenReturn(fakeTable); final ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class); doAnswer( new Answer<Job>() { @Override public Job answer(InvocationOnMock invocationOnMock) throws Throwable { verify(mockBigqueryJobs, times(1)).insert(eq(jobProjectId), jobCaptor.capture()); return jobCaptor.getValue(); } }) .when(mockBigqueryJobsInsert) .execute(); when(mockBigqueryJobsGet.execute()).thenReturn(jobHandle); helper.exportBigQueryToGcs(jobProjectId, tableRef, ImmutableList.of("test-export-path"), true); verify(mockBigquery, times(2)).jobs(); verify(mockBigqueryJobs, times(1)) .get(eq(jobProjectId), eq(jobCaptor.getValue().getJobReference().getJobId())); Job job = jobCaptor.getValue(); assertThat(job.getConfiguration().getExtract().getDestinationUris().get(0)) .isEqualTo("test-export-path"); assertThat(job.getConfiguration().getExtract().getSourceTable()).isEqualTo(tableRef); assertThat(job.getJobReference().getLocation()).isEqualTo("test_location"); verify(mockBigqueryJobsGet).setLocation(eq("test_location")); verify(mockBigqueryJobsGet, times(1)).execute(); verify(mockBigqueryJobsInsert, times(1)).execute(); verify(mockBigquery).tables(); verify(mockBigqueryTables).get(eq(projectId), eq(datasetId), eq(tableId)); verify(mockBigqueryTablesGet).execute(); }
public void exportBigQueryToGcs( String projectId, TableReference tableRef, List<String> gcsPaths, boolean awaitCompletion) throws IOException, InterruptedException { logger.atFine().log( "exportBigQueryToGcs(bigquery, '%s', '%s', '%s', '%s')", projectId, BigQueryStrings.toString(tableRef), gcsPaths, awaitCompletion); logger.atInfo().log( "Exporting table '%s' to %s paths; path[0] is '%s'; awaitCompletion: %s", BigQueryStrings.toString(tableRef), gcsPaths.size(), gcsPaths.isEmpty() ? "(empty)" : gcsPaths.get(0), awaitCompletion); JobConfigurationExtract extractConfig = new JobConfigurationExtract(); extractConfig.setSourceTable(tableRef); extractConfig.setDestinationUris(gcsPaths); extractConfig.set("destinationFormat", "NEWLINE_DELIMITED_JSON"); JobConfiguration config = new JobConfiguration(); config.setExtract(extractConfig); Table table = getTable(tableRef); JobReference jobReference = createJobReference(projectId, "direct-bigqueryhelper-export", table.getLocation()); Job job = new Job(); job.setConfiguration(config); job.setJobReference(jobReference); insertJobOrFetchDuplicate(projectId, job); if (awaitCompletion) { BigQueryUtils.waitForJobCompletion(service, projectId, jobReference, () -> {}); } }
BigQueryHelper { public void exportBigQueryToGcs( String projectId, TableReference tableRef, List<String> gcsPaths, boolean awaitCompletion) throws IOException, InterruptedException { logger.atFine().log( "exportBigQueryToGcs(bigquery, '%s', '%s', '%s', '%s')", projectId, BigQueryStrings.toString(tableRef), gcsPaths, awaitCompletion); logger.atInfo().log( "Exporting table '%s' to %s paths; path[0] is '%s'; awaitCompletion: %s", BigQueryStrings.toString(tableRef), gcsPaths.size(), gcsPaths.isEmpty() ? "(empty)" : gcsPaths.get(0), awaitCompletion); JobConfigurationExtract extractConfig = new JobConfigurationExtract(); extractConfig.setSourceTable(tableRef); extractConfig.setDestinationUris(gcsPaths); extractConfig.set("destinationFormat", "NEWLINE_DELIMITED_JSON"); JobConfiguration config = new JobConfiguration(); config.setExtract(extractConfig); Table table = getTable(tableRef); JobReference jobReference = createJobReference(projectId, "direct-bigqueryhelper-export", table.getLocation()); Job job = new Job(); job.setConfiguration(config); job.setJobReference(jobReference); insertJobOrFetchDuplicate(projectId, job); if (awaitCompletion) { BigQueryUtils.waitForJobCompletion(service, projectId, jobReference, () -> {}); } } }
BigQueryHelper { public void exportBigQueryToGcs( String projectId, TableReference tableRef, List<String> gcsPaths, boolean awaitCompletion) throws IOException, InterruptedException { logger.atFine().log( "exportBigQueryToGcs(bigquery, '%s', '%s', '%s', '%s')", projectId, BigQueryStrings.toString(tableRef), gcsPaths, awaitCompletion); logger.atInfo().log( "Exporting table '%s' to %s paths; path[0] is '%s'; awaitCompletion: %s", BigQueryStrings.toString(tableRef), gcsPaths.size(), gcsPaths.isEmpty() ? "(empty)" : gcsPaths.get(0), awaitCompletion); JobConfigurationExtract extractConfig = new JobConfigurationExtract(); extractConfig.setSourceTable(tableRef); extractConfig.setDestinationUris(gcsPaths); extractConfig.set("destinationFormat", "NEWLINE_DELIMITED_JSON"); JobConfiguration config = new JobConfiguration(); config.setExtract(extractConfig); Table table = getTable(tableRef); JobReference jobReference = createJobReference(projectId, "direct-bigqueryhelper-export", table.getLocation()); Job job = new Job(); job.setConfiguration(config); job.setJobReference(jobReference); insertJobOrFetchDuplicate(projectId, job); if (awaitCompletion) { BigQueryUtils.waitForJobCompletion(service, projectId, jobReference, () -> {}); } } BigQueryHelper(Bigquery service); }
BigQueryHelper { public void exportBigQueryToGcs( String projectId, TableReference tableRef, List<String> gcsPaths, boolean awaitCompletion) throws IOException, InterruptedException { logger.atFine().log( "exportBigQueryToGcs(bigquery, '%s', '%s', '%s', '%s')", projectId, BigQueryStrings.toString(tableRef), gcsPaths, awaitCompletion); logger.atInfo().log( "Exporting table '%s' to %s paths; path[0] is '%s'; awaitCompletion: %s", BigQueryStrings.toString(tableRef), gcsPaths.size(), gcsPaths.isEmpty() ? "(empty)" : gcsPaths.get(0), awaitCompletion); JobConfigurationExtract extractConfig = new JobConfigurationExtract(); extractConfig.setSourceTable(tableRef); extractConfig.setDestinationUris(gcsPaths); extractConfig.set("destinationFormat", "NEWLINE_DELIMITED_JSON"); JobConfiguration config = new JobConfiguration(); config.setExtract(extractConfig); Table table = getTable(tableRef); JobReference jobReference = createJobReference(projectId, "direct-bigqueryhelper-export", table.getLocation()); Job job = new Job(); job.setConfiguration(config); job.setJobReference(jobReference); insertJobOrFetchDuplicate(projectId, job); if (awaitCompletion) { BigQueryUtils.waitForJobCompletion(service, projectId, jobReference, () -> {}); } } BigQueryHelper(Bigquery service); Bigquery getRawBigquery(); void importFederatedFromGcs(
      String projectId,
      TableReference tableRef,
      @Nullable TableSchema schema,
      BigQueryFileFormat sourceFormat,
      List<String> gcsPaths); void importFromGcs(
      String projectId,
      TableReference tableRef,
      @Nullable TableSchema schema,
      @Nullable TimePartitioning timePartitioning,
      @Nullable String kmsKeyName,
      BigQueryFileFormat sourceFormat,
      String createDisposition,
      String writeDisposition,
      List<String> gcsPaths,
      boolean awaitCompletion); void exportBigQueryToGcs(
      String projectId, TableReference tableRef, List<String> gcsPaths, boolean awaitCompletion); boolean tableExists(TableReference tableRef); Table getTable(TableReference tableRef); JobReference createJobReference(
      String projectId, String jobIdPrefix, @Nullable String location); void checkJobIdEquality(Job expected, Job actual); Job insertJobOrFetchDuplicate(String projectId, Job job); }
BigQueryHelper { public void exportBigQueryToGcs( String projectId, TableReference tableRef, List<String> gcsPaths, boolean awaitCompletion) throws IOException, InterruptedException { logger.atFine().log( "exportBigQueryToGcs(bigquery, '%s', '%s', '%s', '%s')", projectId, BigQueryStrings.toString(tableRef), gcsPaths, awaitCompletion); logger.atInfo().log( "Exporting table '%s' to %s paths; path[0] is '%s'; awaitCompletion: %s", BigQueryStrings.toString(tableRef), gcsPaths.size(), gcsPaths.isEmpty() ? "(empty)" : gcsPaths.get(0), awaitCompletion); JobConfigurationExtract extractConfig = new JobConfigurationExtract(); extractConfig.setSourceTable(tableRef); extractConfig.setDestinationUris(gcsPaths); extractConfig.set("destinationFormat", "NEWLINE_DELIMITED_JSON"); JobConfiguration config = new JobConfiguration(); config.setExtract(extractConfig); Table table = getTable(tableRef); JobReference jobReference = createJobReference(projectId, "direct-bigqueryhelper-export", table.getLocation()); Job job = new Job(); job.setConfiguration(config); job.setJobReference(jobReference); insertJobOrFetchDuplicate(projectId, job); if (awaitCompletion) { BigQueryUtils.waitForJobCompletion(service, projectId, jobReference, () -> {}); } } BigQueryHelper(Bigquery service); Bigquery getRawBigquery(); void importFederatedFromGcs(
      String projectId,
      TableReference tableRef,
      @Nullable TableSchema schema,
      BigQueryFileFormat sourceFormat,
      List<String> gcsPaths); void importFromGcs(
      String projectId,
      TableReference tableRef,
      @Nullable TableSchema schema,
      @Nullable TimePartitioning timePartitioning,
      @Nullable String kmsKeyName,
      BigQueryFileFormat sourceFormat,
      String createDisposition,
      String writeDisposition,
      List<String> gcsPaths,
      boolean awaitCompletion); void exportBigQueryToGcs(
      String projectId, TableReference tableRef, List<String> gcsPaths, boolean awaitCompletion); boolean tableExists(TableReference tableRef); Table getTable(TableReference tableRef); JobReference createJobReference(
      String projectId, String jobIdPrefix, @Nullable String location); void checkJobIdEquality(Job expected, Job actual); Job insertJobOrFetchDuplicate(String projectId, Job job); static final String BIGQUERY_JOB_ID_PATTERN; static final int BIGQUERY_JOB_ID_MAX_LENGTH; }