import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
@Override
public void connect() {
    LOG.info("Connecting to BigQuery");
    // Instantiates a client
    bigquery = BigQueryOptions.getDefaultInstance().getService(); 
 
    // check to see if the dataset already exists
    dataset = bigquery.getDataset(datasetName);
    if (dataset == null) {
        // dataset was not found, so create it
        LOG.info("Dataset {} was not found. Creating it ...", datasetName);
        
        // Prepares a new dataset
        datasetInfo = DatasetInfo.newBuilder(datasetName).build();

        // Creates the dataset.
        dataset = bigquery.create(datasetInfo);
        
        LOG.info("Dataset {} created.", dataset.getDatasetId().getDataset());
    } else {
        LOG.info("Dataset {} found.", dataset.getDatasetId().getDataset());
    }
    
    logBigQueryInfo();
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
/**
 * Returns a default {@link BigQuery} instance for the specified project with credentials provided
 * in the specified file, which can then be used for creating, updating, and inserting into tables
 * from specific datasets.
 *
 * @param projectName The name of the BigQuery project to work with
 * @param keyFilename The name of a file containing a JSON key that can be used to provide
 *                    credentials to BigQuery, or null if no authentication should be performed.
 * @return The resulting BigQuery object.
 */
public BigQuery connect(String projectName, String keyFilename) {
  if (keyFilename == null) {
    return connect(projectName);
  }

  logger.debug("Attempting to open file {} for service account json key", keyFilename);
  try (InputStream credentialsStream = new FileInputStream(keyFilename)) {
    logger.debug("Attempting to authenticate with BigQuery using provided json key");
    return new BigQueryOptions.DefaultBigqueryFactory().create(
        BigQueryOptions.newBuilder()
        .setProjectId(projectName)
        .setCredentials(GoogleCredentials.fromStream(credentialsStream))
        .build()
    );
  } catch (IOException err) {
    throw new BigQueryConnectException("Failed to access json key file", err);
  }
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
public static void explicit() throws IOException {
  // Load credentials from JSON key file. If you can't set the GOOGLE_APPLICATION_CREDENTIALS
  // environment variable, you can explicitly load the credentials file to construct the
  // credentials.
  GoogleCredentials credentials;
  File credentialsPath = new File("service_account.json");  // TODO: update to your key path.
  try (FileInputStream serviceAccountStream = new FileInputStream(credentialsPath)) {
    credentials = ServiceAccountCredentials.fromStream(serviceAccountStream);
  }

  // Instantiate a client.
  BigQuery bigquery =
      BigQueryOptions.newBuilder().setCredentials(credentials).build().getService();

  // Use the client.
  System.out.println("Datasets:");
  for (Dataset dataset : bigquery.listDatasets().iterateAll()) {
    System.out.printf("%s%n", dataset.getDatasetId().getDataset());
  }
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
public static void main(String... args) throws Exception {
  // Instantiate a client. If you don't specify credentials when constructing a client, the
  // client library will look for credentials in the environment, such as the
  // GOOGLE_APPLICATION_CREDENTIALS environment variable.
  BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();

  // The name for the new dataset
  String datasetName = "my_new_dataset";

  // Prepares a new dataset
  Dataset dataset = null;
  DatasetInfo datasetInfo = DatasetInfo.newBuilder(datasetName).build();

  // Creates the dataset
  dataset = bigquery.create(datasetInfo);

  System.out.printf("Dataset %s created.%n", dataset.getDatasetId().getDataset());
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
public static void runQuery(QueryJobConfiguration queryConfig)
    throws TimeoutException, InterruptedException {
  BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();

  // Create a job ID so that we can safely retry.
  JobId jobId = JobId.of(UUID.randomUUID().toString());
  Job queryJob = bigquery.create(JobInfo.newBuilder(queryConfig).setJobId(jobId).build());

  // Wait for the query to complete.
  queryJob = queryJob.waitFor();

  // Check for errors
  if (queryJob == null) {
    throw new RuntimeException("Job no longer exists");
  } else if (queryJob.getStatus().getError() != null) {
    // You can also look at queryJob.getStatus().getExecutionErrors() for all
    // errors, not just the latest one.
    throw new RuntimeException(queryJob.getStatus().getError().toString());
  }

  // Get the results.
  QueryResponse response = bigquery.getQueryResults(jobId);
  QueryResult result = response.getResult();

  // Print all pages of the results.
  while (result != null) {
    for (List<FieldValue> row : result.iterateAll()) {
      for (FieldValue val : row) {
        System.out.printf("%s,", val.toString());
      }
      System.out.printf("\n");
    }

    result = result.getNextPage();
  }
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
public static void implicit() {
  // Instantiate a client. If you don't specify credentials when constructing a client, the
  // client library will look for credentials in the environment, such as the
  // GOOGLE_APPLICATION_CREDENTIALS environment variable.
  BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();

  // Use the client.
  System.out.println("Datasets:");
  for (Dataset dataset : bigquery.listDatasets().iterateAll()) {
    System.out.printf("%s%n", dataset.getDatasetId().getDataset());
  }
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
private static final void deleteMyNewDataset() {
  BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
  String datasetName = "my_new_dataset";
  DatasetId datasetId = DatasetId.of(datasetName);
  DatasetDeleteOption deleteContents = DatasetDeleteOption.deleteContents();
  bigquery.delete(datasetId, deleteContents);
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
public static BigQuery getBigquery(Credentials credentials, String projectId) {
  BigQueryOptions options = BigQueryOptions.newBuilder()
      .setCredentials(credentials)
      .setProjectId(projectId)
      .build();
  return options.getService();
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
public static BigQuery createClient(BigQueryDatastoreProperties datastore) {
    if (StringUtils.isEmpty(datastore.serviceAccountFile.getValue())) {
        return BigQueryOptions.getDefaultInstance().getService();
    } else {
        return BigQueryOptions.newBuilder().setProjectId(datastore.projectName.getValue())
                .setCredentials(createCredentials(datastore)).build().getService();
    }
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
public static void main( String[] args )
{
	// Step 1: Generate json data file
	System.out.println("Step 1 start: Generate json data file at: " + (new Date()).toString());
	String jsonFile = writeSalesOrder(1000000); // 1 million records
	System.out.println("Step 1 end: Generate json data file at: " + (new Date()).toString());
	
	// Step 2: Compress json data file
	System.out.println("\nStep 2 start: Compress json data file at: " + (new Date()).toString());
	String gzipFile = GZipHelper.compressGzipFile(jsonFile, jsonFile + ".gz");
	System.out.println("Step 2 end: Compress json data file end at: " + (new Date()).toString());
	
	// Step 3: Get BigQuery instance
	System.out.println("\nStep 3 start: Get BigQuery instance at: " + (new Date()).toString());
	BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
	BigQuerySnippets snippets = new BigQuerySnippets(bigquery);
	System.out.println("Step 3 end: Get BigQuery instance at: " + (new Date()).toString());
	
	// Step 4: Create dataset and table in google bigquery
	System.out.println("\nStep 4 start: Create dataset and table in google bigquery at: " + (new Date()).toString());
	String datasetName = "sales_order_ds";
	String tableName = "sales_order";
	boolean isAppendData = false;
	try{
		Dataset dataset = snippets.getDataset(datasetName);
    	if(dataset == null)
    		snippets.createDataset(datasetName);
    	Table table = bigquery.getTable(datasetName, tableName);
    	if(table == null){
    		snippets.createTable(datasetName, tableName, snippets.createSchema());
    	}else{
    		if(isAppendData == false){
    			// program will delete the existing table first, then create a new table with the same schema
     		Schema schema = table.getDefinition().getSchema();
     		Boolean deleteSuccess = snippets.deleteTable(datasetName, tableName);
     		if(deleteSuccess)
     			snippets.createTable(datasetName, tableName, schema);
    		}
    	}
    	System.out.println("DataSet: " + datasetName + ", table: " + tableName + " were created!");
	}catch(Exception e){
		e.printStackTrace();
	}
	System.out.println("Step 4 end: Create dataset and table in google bigquery at: " + (new Date()).toString());
	
	// Step 5: Loading json.gz file into google bigquery
	System.out.println("\nStep 5 start: Loading json.gz file into google bigquery at: " + (new Date()).toString());
	loadSalesOrder(gzipFile, bigquery,snippets, datasetName, tableName);
	System.out.println("Step 5 end: Loading json.gz file into google bigquery at: " + (new Date()).toString());
	
	// Step 6: Executing sql query command
	System.out.println("\nStep 6 start: Executing sql query command at: " + (new Date()).toString());
	String sql = "SELECT product_name, sum(product_price) total_amount" +
				 " FROM [bigquery-study-157507:" + datasetName + "." + tableName + "]" + // From project_name.datasetName.tableName
				 " group by product_name" +
				 " order by total_amount desc" +
				 " LIMIT 10";
	execBigQuery(sql, snippets);
	System.out.println("Step 6 end: Executing sql query command at: " + (new Date()).toString());
	
	System.out.println("\nCelebrating to yourself!!!");	
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
public static void main(String... args) throws Exception {
  // [START bigquery_simple_app_client]
  BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
  // [END bigquery_simple_app_client]
  // [START bigquery_simple_app_query]
  QueryJobConfiguration queryConfig =
      QueryJobConfiguration.newBuilder(
        "SELECT "
            + "CONCAT('https://stackoverflow.com/questions/', CAST(id as STRING)) as url, "
            + "view_count "
            + "FROM `bigquery-public-data.stackoverflow.posts_questions` "
            + "WHERE tags like '%google-bigquery%' "
            + "ORDER BY favorite_count DESC LIMIT 10")
          // Use standard SQL syntax for queries.
          // See: https://cloud.google.com/bigquery/sql-reference/
          .setUseLegacySql(false)
          .build();

  // Create a job ID so that we can safely retry.
  JobId jobId = JobId.of(UUID.randomUUID().toString());
  Job queryJob = bigquery.create(JobInfo.newBuilder(queryConfig).setJobId(jobId).build());

  // Wait for the query to complete.
  queryJob = queryJob.waitFor();

  // Check for errors
  if (queryJob == null) {
    throw new RuntimeException("Job no longer exists");
  } else if (queryJob.getStatus().getError() != null) {
    // You can also look at queryJob.getStatus().getExecutionErrors() for all
    // errors, not just the latest one.
    throw new RuntimeException(queryJob.getStatus().getError().toString());
  }
  // [END bigquery_simple_app_query]

  // [START bigquery_simple_app_print]
  // Get the results.
  QueryResponse response = bigquery.getQueryResults(jobId);

  QueryResult result = response.getResult();

  // Print all pages of the results.
  for (FieldValueList row : result.iterateAll()) {
    String url = row.get("url").getStringValue();
    long viewCount = row.get("view_count").getLongValue();
    System.out.printf("url: %s views: %d%n", url, viewCount);
  }
  // [END bigquery_simple_app_print]
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
/**
 * Query the Shakespeare dataset for words with count at least {@code minWordCount} in the corpus
 * {@code corpus}.
 */
// [START bigquery_query_params]
private static void runNamed(final String corpus, final long minWordCount)
    throws InterruptedException {
  BigQuery bigquery =
      new BigQueryOptions.DefaultBigqueryFactory().create(BigQueryOptions.getDefaultInstance());

  String queryString =
      "SELECT word, word_count\n"
          + "FROM `bigquery-public-data.samples.shakespeare`\n"
          + "WHERE corpus = @corpus\n"
          + "AND word_count >= @min_word_count\n"
          + "ORDER BY word_count DESC";
  QueryJobConfiguration queryRequest =
      QueryJobConfiguration.newBuilder(queryString)
          .addNamedParameter("corpus", QueryParameterValue.string(corpus))
          .addNamedParameter("min_word_count", QueryParameterValue.int64(minWordCount))
          // Standard SQL syntax is required for parameterized queries.
          // See: https://cloud.google.com/bigquery/sql-reference/
          .setUseLegacySql(false)
          .build();

  // Execute the query.
  QueryResponse response = bigquery.query(queryRequest);

  // Wait for the job to finish (if the query takes more than 10 seconds to complete).
  while (!response.jobCompleted()) {
    Thread.sleep(1000);
    response = bigquery.getQueryResults(response.getJobId());
  }

  // Check for errors.
  if (response.hasErrors()) {
    String firstError = "";
    if (response.getExecutionErrors().size() != 0) {
      firstError = response.getExecutionErrors().get(0).getMessage();
    }
    throw new RuntimeException(firstError);
  }

  // Print all pages of the results.
  QueryResult result = response.getResult();
  while (result != null) {
    for (List<FieldValue> row : result.iterateAll()) {
      System.out.printf("%s: %d\n", row.get(0).getStringValue(), row.get(1).getLongValue());
    }

    result = result.getNextPage();
  }
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
/**
 * Query the baby names database to find the most popular names for a gender in a list of states.
 */
// [START bigquery_query_params_arrays]
private static void runArray(String gender, String[] states) throws InterruptedException {
  BigQuery bigquery =
      new BigQueryOptions.DefaultBigqueryFactory().create(BigQueryOptions.getDefaultInstance());

  String queryString =
      "SELECT name, sum(number) as count\n"
          + "FROM `bigquery-public-data.usa_names.usa_1910_2013`\n"
          + "WHERE gender = @gender\n"
          + "AND state IN UNNEST(@states)\n"
          + "GROUP BY name\n"
          + "ORDER BY count DESC\n"
          + "LIMIT 10;";
  QueryJobConfiguration queryRequest =
      QueryJobConfiguration.newBuilder(queryString)
          .addNamedParameter("gender", QueryParameterValue.string(gender))
          .addNamedParameter("states", QueryParameterValue.array(states, String.class))
          // Standard SQL syntax is required for parameterized queries.
          // See: https://cloud.google.com/bigquery/sql-reference/
          .setUseLegacySql(false)
          .build();

  // Execute the query.
  QueryResponse response = bigquery.query(queryRequest);

  // Wait for the job to finish (if the query takes more than 10 seconds to complete).
  while (!response.jobCompleted()) {
    Thread.sleep(1000);
    response = bigquery.getQueryResults(response.getJobId());
  }

  // Check for errors.
  if (response.hasErrors()) {
    String firstError = "";
    if (response.getExecutionErrors().size() != 0) {
      firstError = response.getExecutionErrors().get(0).getMessage();
    }
    throw new RuntimeException(firstError);
  }

  // Print all pages of the results.
  QueryResult result = response.getResult();
  while (result != null) {
    for (List<FieldValue> row : result.iterateAll()) {
      System.out.printf("%s: %d\n", row.get(0).getStringValue(), row.get(1).getLongValue());
    }

    result = result.getNextPage();
  }
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
private static void runTimestamp() throws InterruptedException {
  BigQuery bigquery =
      new BigQueryOptions.DefaultBigqueryFactory().create(BigQueryOptions.getDefaultInstance());

  DateTime timestamp = new DateTime(2016, 12, 7, 8, 0, 0, DateTimeZone.UTC);

  String queryString = "SELECT TIMESTAMP_ADD(@ts_value, INTERVAL 1 HOUR);";
  QueryJobConfiguration queryRequest =
      QueryJobConfiguration.newBuilder(queryString)
          .addNamedParameter(
              "ts_value",
              QueryParameterValue.timestamp(
                  // Timestamp takes microseconds since 1970-01-01T00:00:00 UTC
                  timestamp.getMillis() * 1000))
          // Standard SQL syntax is required for parameterized queries.
          // See: https://cloud.google.com/bigquery/sql-reference/
          .setUseLegacySql(false)
          .build();

  // Execute the query.
  QueryResponse response = bigquery.query(queryRequest);

  // Wait for the job to finish (if the query takes more than 10 seconds to complete).
  while (!response.jobCompleted()) {
    Thread.sleep(1000);
    response = bigquery.getQueryResults(response.getJobId());
  }

  // Check for errors.
  if (response.hasErrors()) {
    String firstError = "";
    if (response.getExecutionErrors().size() != 0) {
      firstError = response.getExecutionErrors().get(0).getMessage();
    }
    throw new RuntimeException(firstError);
  }

  // Print all pages of the results.
  QueryResult result = response.getResult();
  DateTimeFormatter formatter = ISODateTimeFormat.dateTimeNoMillis().withZoneUTC();
  while (result != null) {
    for (List<FieldValue> row : result.iterateAll()) {
      System.out.printf(
          "%s\n",
          formatter.print(
              new DateTime(
                  // Timestamp values are returned in microseconds since 1970-01-01T00:00:00 UTC,
                  // but org.joda.time.DateTime constructor accepts times in milliseconds.
                  row.get(0).getTimestampValue() / 1000, DateTimeZone.UTC)));
    }

    result = result.getNextPage();
  }
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
private static final void deleteTestDataset() {
  BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
  DatasetId datasetId = DatasetId.of(TEST_DATASET);
  BigQuery.DatasetDeleteOption deleteContents = BigQuery.DatasetDeleteOption.deleteContents();
  bigquery.delete(datasetId, deleteContents);
}
 

import com.google.cloud.bigquery.BigQueryOptions; //导入依赖的package包/类
private static final void createTestDataset() {
  BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
  DatasetId datasetId = DatasetId.of(TEST_DATASET);
  bigquery.create(DatasetInfo.newBuilder(datasetId).build());
}