import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
private void setupBigQueryTable(String projectId, String datasetId, String tableId,
    TableSchema schema) throws IOException {
  if (bigQueryClient == null) {
    bigQueryClient = Transport.newBigQueryClient(options.as(BigQueryOptions.class)).build();
  }

  Datasets datasetService = bigQueryClient.datasets();
  if (executeNullIfNotFound(datasetService.get(projectId, datasetId)) == null) {
    Dataset newDataset = new Dataset().setDatasetReference(
        new DatasetReference().setProjectId(projectId).setDatasetId(datasetId));
    datasetService.insert(projectId, newDataset).execute();
  }

  Tables tableService = bigQueryClient.tables();
  Table table = executeNullIfNotFound(tableService.get(projectId, datasetId, tableId));
  if (table == null) {
    Table newTable = new Table().setSchema(schema).setTableReference(
        new TableReference().setProjectId(projectId).setDatasetId(datasetId).setTableId(tableId));
    tableService.insert(projectId, datasetId, newTable).execute();
  } else if (!table.getSchema().equals(schema)) {
    throw new RuntimeException(
        "Table exists and schemas do not match, expecting: " + schema.toPrettyString()
        + ", actual: " + table.getSchema().toPrettyString());
  }
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
private void setupBigQueryTable(String projectId, String datasetId, String tableId,
    TableSchema schema) throws IOException {
  if (bigQueryClient == null) {
    bigQueryClient = newBigQueryClient(options.as(BigQueryOptions.class)).build();
  }

  Datasets datasetService = bigQueryClient.datasets();
  if (executeNullIfNotFound(datasetService.get(projectId, datasetId)) == null) {
    Dataset newDataset = new Dataset().setDatasetReference(
        new DatasetReference().setProjectId(projectId).setDatasetId(datasetId));
    datasetService.insert(projectId, newDataset).execute();
  }

  Tables tableService = bigQueryClient.tables();
  Table table = executeNullIfNotFound(tableService.get(projectId, datasetId, tableId));
  if (table == null) {
    Table newTable = new Table().setSchema(schema).setTableReference(
        new TableReference().setProjectId(projectId).setDatasetId(datasetId).setTableId(tableId));
    tableService.insert(projectId, datasetId, newTable).execute();
  } else if (!table.getSchema().equals(schema)) {
    throw new RuntimeException(
        "Table exists and schemas do not match, expecting: " + schema.toPrettyString()
        + ", actual: " + table.getSchema().toPrettyString());
  }
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
/**
 * Ensures the dataset exists by trying to create it. Note that it's not appreciably cheaper
 * to check for dataset existence than it is to try to create it and check for exceptions.
 */
// Note that these are not static so they can be mocked for testing.
private void ensureDataset(Bigquery bigquery, String projectId, String datasetId)
    throws IOException {
  try {
    bigquery.datasets()
        .insert(projectId,
            new Dataset().setDatasetReference(
                new DatasetReference()
                    .setProjectId(projectId)
                    .setDatasetId(datasetId)))
        .execute();
  } catch (IOException e) {
    // Swallow errors about a duplicate dataset, and throw any other ones.
    if (!BigqueryJobFailureException.create(e).getReason().equals("duplicate")) {
      throw e;
    }
  }
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
@Before
public void before() throws Exception {
  when(subfactory.create(
      anyString(),
      any(HttpTransport.class),
      any(JsonFactory.class),
      any(HttpRequestInitializer.class)))
          .thenReturn(bigquery);
  when(bigquery.datasets()).thenReturn(bigqueryDatasets);
  when(bigqueryDatasets.insert(eq("Project-Id"), any(Dataset.class)))
      .thenReturn(bigqueryDatasetsInsert);
  when(bigquery.tables()).thenReturn(bigqueryTables);
  when(bigqueryTables.insert(eq("Project-Id"), any(String.class), any(Table.class)))
      .thenReturn(bigqueryTablesInsert);
  factory = new BigqueryFactory();
  factory.subfactory = subfactory;
  factory.bigquerySchemas =
      new ImmutableMap.Builder<String, ImmutableList<TableFieldSchema>>()
          .put(
              "Table-Id",
              ImmutableList.of(new TableFieldSchema().setName("column1").setType(STRING.name())))
          .put(
              "Table2",
              ImmutableList.of(new TableFieldSchema().setName("column1").setType(STRING.name())))
          .build();
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
@Test
public void testSuccess_datastoreAndTableCreation() throws Exception {
  factory.create("Project-Id", "Dataset2", "Table2");

  ArgumentCaptor<Dataset> datasetArg = ArgumentCaptor.forClass(Dataset.class);
  verify(bigqueryDatasets).insert(eq("Project-Id"), datasetArg.capture());
  assertThat(datasetArg.getValue().getDatasetReference().getProjectId())
      .isEqualTo("Project-Id");
  assertThat(datasetArg.getValue().getDatasetReference().getDatasetId())
      .isEqualTo("Dataset2");
  verify(bigqueryDatasetsInsert).execute();

  ArgumentCaptor<Table> tableArg = ArgumentCaptor.forClass(Table.class);
  verify(bigqueryTables).insert(eq("Project-Id"), eq("Dataset2"), tableArg.capture());
  TableReference ref = tableArg.getValue().getTableReference();
  assertThat(ref.getProjectId()).isEqualTo("Project-Id");
  assertThat(ref.getDatasetId()).isEqualTo("Dataset2");
  assertThat(ref.getTableId()).isEqualTo("Table2");
  assertThat(tableArg.getValue().getSchema().getFields())
      .containsExactly(new TableFieldSchema().setName("column1").setType(STRING.name()));
  verify(bigqueryTablesInsert).execute();
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
@Before
public void before() throws Exception {
  when(bigqueryFactory.create("Project-Id", "snapshots")).thenReturn(bigquery);
  when(bigquery.jobs()).thenReturn(bigqueryJobs);
  when(bigqueryJobs.insert(eq("Project-Id"), any(Job.class))).thenReturn(bigqueryJobsInsert);
  when(bigquery.datasets()).thenReturn(bigqueryDatasets);
  when(bigqueryDatasets.insert(eq("Project-Id"), any(Dataset.class)))
      .thenReturn(bigqueryDatasetsInsert);
  action = new LoadSnapshotAction();
  action.bigqueryFactory = bigqueryFactory;
  action.bigqueryPollEnqueuer = bigqueryPollEnqueuer;
  action.clock = clock;
  action.projectId = "Project-Id";
  action.snapshotFile = "gs://bucket/snapshot.backup_info";
  action.snapshotId = "id12345";
  action.snapshotKinds = "one,two,three";
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
@Before
public void before() throws Exception {
  when(bigqueryFactory.create(anyString(), anyString())).thenReturn(bigquery);
  when(bigquery.datasets()).thenReturn(bigqueryDatasets);
  when(bigqueryDatasets.insert(anyString(), any(Dataset.class)))
      .thenReturn(bigqueryDatasetsInsert);
  when(bigquery.tables()).thenReturn(bigqueryTables);
  when(bigqueryTables.update(anyString(), anyString(), anyString(), any(Table.class)))
      .thenReturn(bigqueryTablesUpdate);

  action = new UpdateSnapshotViewAction();
  action.bigqueryFactory = bigqueryFactory;
  action.datasetId = "some_dataset";
  action.kindName = "fookind";
  action.projectId = "myproject";
  action.tableId = "12345_fookind";
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
void createDataset(String projectId, Dataset dataset)
        throws IOException
{
    try {
        client.datasets().insert(projectId, dataset)
                .execute();
    }
    catch (GoogleJsonResponseException e) {
        if (e.getStatusCode() == HttpStatusCodes.STATUS_CODE_CONFLICT) {
            logger.debug("Dataset already exists: {}:{}", dataset.getDatasetReference());
        }
        else {
            throw e;
        }
    }
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
private Dataset dataset(String defaultProjectId, JsonNode node)
{
    if (node.isTextual()) {
        return new Dataset()
                .setDatasetReference(datasetReference(defaultProjectId, node.asText()));
    }
    else {
        DatasetConfig config;
        try {
            config = objectMapper.readValue(node.traverse(), DatasetConfig.class);
        }
        catch (IOException e) {
            throw new ConfigException("Invalid dataset reference or configuration: " + node, e);
        }
        return dataset(defaultProjectId, config);
    }
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
/**
 * Tests the setupJob method of BigQueryOutputFormat.
 */
@Test
public void testSetupJob() 
    throws IOException {
  // Mock method calls.
  when(mockBigquery.datasets()).thenReturn(mockBigqueryDatasets);
  when(mockBigqueryDatasets.insert(any(String.class), any(Dataset.class)))
      .thenReturn(mockBigqueryDatasetsInsert);

  // Run method and verify calls.
  committerInstance.setupJob(jobContext);
  verify(mockBigquery).datasets();
  verify(mockBigqueryDatasets).insert(eq(TEMP_PROJECT_ID), eq(expectedTempDataset));
  verify(mockBigqueryDatasetsInsert, times(1)).execute();
  verify(mockBigQueryHelper, atLeastOnce()).getRawBigquery();
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
/**
 * {@inheritDoc}
 *
 * <p>Tries executing the RPC for at most {@code MAX_RPC_RETRIES} times until it succeeds.
 *
 * @throws IOException if it exceeds {@code MAX_RPC_RETRIES} attempts.
 */
@Override
public Dataset getDataset(String projectId, String datasetId)
    throws IOException, InterruptedException {
  return executeWithRetries(
      client.datasets().get(projectId, datasetId),
      String.format(
          "Unable to get dataset: %s, aborting after %d retries.",
          datasetId, MAX_RPC_RETRIES),
      Sleeper.DEFAULT,
      createDefaultBackoff(),
      DONT_RETRY_NOT_FOUND);
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
@Override
public Dataset getDataset(
    String projectId, String datasetId) throws IOException, InterruptedException {
  synchronized (tables) {
    Map<String, TableContainer> dataset = tables.get(projectId, datasetId);
    if (dataset == null) {
      throwNotFound("Tried to get a dataset %s:%s, but no such table was set",
                  projectId, datasetId);
    }
    return new Dataset().setDatasetReference(new DatasetReference()
        .setDatasetId(datasetId)
        .setProjectId(projectId));
  }
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
/**
 * Helper that creates a dataset with this name if it doesn't already exist, and returns true
 * if creation took place.
 */
public boolean createDatasetIfNeeded(String datasetName) throws IOException {
  if (!checkDatasetExists(datasetName)) {
    bigquery.datasets()
        .insert(getProjectId(), new Dataset().setDatasetReference(new DatasetReference()
            .setProjectId(getProjectId())
            .setDatasetId(datasetName)))
        .execute();
    System.err.printf("Created dataset: %s:%s\n", getProjectId(), datasetName);
    return true;
  }
  return false;
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
@Test
public void testSuccess_datastoreCreation() throws Exception {
  factory.create("Project-Id", "Dataset-Id");

  ArgumentCaptor<Dataset> datasetArg = ArgumentCaptor.forClass(Dataset.class);
  verify(bigqueryDatasets).insert(eq("Project-Id"), datasetArg.capture());
  assertThat(datasetArg.getValue().getDatasetReference().getProjectId())
      .isEqualTo("Project-Id");
  assertThat(datasetArg.getValue().getDatasetReference().getDatasetId())
      .isEqualTo("Dataset-Id");
  verify(bigqueryDatasetsInsert).execute();
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
void emptyDataset(String projectId, Dataset dataset)
        throws IOException
{
    String datasetId = dataset.getDatasetReference().getDatasetId();
    deleteDataset(projectId, datasetId);
    createDataset(projectId, dataset);
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
@Test
public void testLoad()
        throws Exception
{
    assumeThat(GCS_TEST_BUCKET, not(isEmptyOrNullString()));

    // Create source data object
    String objectName = GCS_PREFIX + "test.csv";
    byte[] data = Joiner.on('\n').join("a,b", "c,d").getBytes(UTF_8);
    InputStreamContent content = new InputStreamContent("text/csv", new ByteArrayInputStream(data))
            .setLength(data.length);
    StorageObject metadata = new StorageObject().setName(objectName);
    gcs.objects()
            .insert(GCS_TEST_BUCKET, metadata, content)
            .execute();

    // Create output dataset
    String datasetId = BQ_TAG + "_load_test";
    Dataset dataset = new Dataset().setDatasetReference(new DatasetReference()
            .setProjectId(gcpProjectId)
            .setDatasetId(datasetId));
    bq.datasets().insert(gcpProjectId, dataset)
            .execute();

    // Run load
    String tableId = "data";
    addWorkflow(projectDir, "acceptance/bigquery/load.dig");
    Id attemptId = pushAndStart(server.endpoint(), projectDir, "load", ImmutableMap.of(
            "source_bucket", GCS_TEST_BUCKET,
            "source_object", objectName,
            "target_dataset", datasetId,
            "target_table", tableId,
            "outfile", outfile.toString()));
    expect(Duration.ofMinutes(5), attemptSuccess(server.endpoint(), attemptId));
    assertThat(Files.exists(outfile), is(true));

    // Check that destination table was created
    Table destinationTable = bq.tables().get(gcpProjectId, datasetId, tableId).execute();
    assertThat(destinationTable.getTableReference().getTableId(), is(tableId));
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
static Dataset createDataset(Bigquery bq, String projectId, String datasetId)
        throws IOException
{
    Dataset dataset = new Dataset()
            .setDatasetReference(new DatasetReference()
                    .setDatasetId(datasetId));
    Dataset created = createDataset(bq, projectId, dataset);
    assertThat(datasetExists(bq, projectId, datasetId), is(true));
    return created;
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
/**
 * Creates the temporary dataset that will contain all of the task work tables.
 *
 * @param context the job's context.
 * @throws IOException on IO Error.
 */
@Override
public void setupJob(JobContext context) throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("setupJob({})", HadoopToStringUtil.toString(context));
  }
  // Create dataset.
  DatasetReference datasetReference = new DatasetReference();
  datasetReference.setProjectId(tempTableRef.getProjectId());
  datasetReference.setDatasetId(tempTableRef.getDatasetId());

  Configuration config = context.getConfiguration();
  Dataset tempDataset = new Dataset();
  tempDataset.setDatasetReference(datasetReference);
  tempDataset.setLocation(config.get(BigQueryConfiguration.DATA_LOCATION_KEY,
                                     BigQueryConfiguration.DATA_LOCATION_DEFAULT));

  // Insert dataset into Bigquery.
  Bigquery.Datasets datasets = bigQueryHelper.getRawBigquery().datasets();

  // TODO(user): Maybe allow the dataset to exist already instead of throwing 409 here.
  LOG.debug("Creating temporary dataset '{}' for project '{}'",
      tempTableRef.getDatasetId(), tempTableRef.getProjectId());

  // NB: Even though this "insert" makes it look like we can specify a different projectId than
  // the one which owns the dataset, it actually has to match.
  datasets.insert(tempTableRef.getProjectId(), tempDataset).execute();
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
/**
 * Tests the setupTask method of BigQueryOutputFormat.
 */
@Test
public void testSetupTask() 
    throws IOException {
  // Mock method calls.
  when(mockBigquery.datasets()).thenReturn(mockBigqueryDatasets);
  when(mockBigqueryDatasets.insert(any(String.class), any(Dataset.class)))
      .thenReturn(mockBigqueryDatasetsInsert);

  // Run method and verify calls.
  committerInstance.setupTask(mockTaskAttemptContext);
  // Tear down verifies no calls are made.
}
 

import com.google.api.services.bigquery.model.Dataset; //导入依赖的package包/类
public void createDataset(String datasetName) throws GeneralSecurityException, IOException {
	Bigquery bigquery = GoogleServices.getBigqueryServiceDomainWide();
	
	Dataset dataset = new Dataset();
    DatasetReference datasetRef = new DatasetReference();
    datasetRef.setProjectId(AppHelper.getAppId());
    datasetRef.setDatasetId(datasetName);
    dataset.setDatasetReference(datasetRef);
    
    try {
    	bigquery.datasets().insert(AppHelper.getAppId(), dataset).execute();
    } catch (GoogleJsonResponseException ex) {
		if (ex.getStatusCode() != 409) throw ex;
	}
}