import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
@Test(expected = ConnectException.class)
public void testPutWhenPartitioningOnMessageTimeWhenNoTimestampType() {
  final String topic = "test-topic";

  Map<String, String> properties = propertiesFactory.getProperties();
  properties.put(BigQuerySinkConfig.TOPICS_CONFIG, topic);
  properties.put(BigQuerySinkConfig.DATASETS_CONFIG, ".*=scratch");
  properties.put(BigQuerySinkTaskConfig.BIGQUERY_MESSAGE_TIME_PARTITIONING_CONFIG, "true");

  BigQuery bigQuery = mock(BigQuery.class);
  SinkTaskContext sinkTaskContext = mock(SinkTaskContext.class);
  InsertAllResponse insertAllResponse = mock(InsertAllResponse.class);

  when(bigQuery.insertAll(anyObject())).thenReturn(insertAllResponse);
  when(insertAllResponse.hasErrors()).thenReturn(false);

  BigQuerySinkTask testTask = new BigQuerySinkTask(bigQuery, null);
  testTask.initialize(sinkTaskContext);
  testTask.start(properties);

  testTask.put(Collections.singletonList(spoofSinkRecord(topic, "value", "message text", TimestampType.NO_TIMESTAMP_TYPE, null)));
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的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.BigQuery; //导入依赖的package包/类
@Test(expected = BigQueryConnectException.class)
public void testNonAutoCreateTablesFailure() {
  final String dataset = "scratch";
  final String existingTableTopic = "topic-with-existing-table";
  final String nonExistingTableTopic = "topic-without-existing-table";
  final TableId existingTable = TableId.of(dataset, "topic_with_existing_table");
  final TableId nonExistingTable = TableId.of(dataset, "topic_without_existing_table");

  Map<String, String> properties = propertiesFactory.getProperties();
  properties.put(BigQuerySinkConnectorConfig.TABLE_CREATE_CONFIG, "false");
  properties.put(BigQuerySinkConfig.SANITIZE_TOPICS_CONFIG, "true");
  properties.put(BigQuerySinkConfig.DATASETS_CONFIG, String.format(".*=%s", dataset));
  properties.put(
      BigQuerySinkConfig.TOPICS_CONFIG,
      String.format("%s, %s", existingTableTopic, nonExistingTableTopic)
  );

  BigQuery bigQuery = mock(BigQuery.class);
  Table fakeTable = mock(Table.class);
  when(bigQuery.getTable(existingTable)).thenReturn(fakeTable);
  when(bigQuery.getTable(nonExistingTable)).thenReturn(null);

  BigQuerySinkConnector testConnector = new BigQuerySinkConnector(bigQuery);
  testConnector.start(properties);
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
@Test
public void testSimplePut() {
  final String topic = "test-topic";

  Map<String, String> properties = propertiesFactory.getProperties();
  properties.put(BigQuerySinkConfig.TOPICS_CONFIG, topic);
  properties.put(BigQuerySinkConfig.DATASETS_CONFIG, ".*=scratch");

  BigQuery bigQuery = mock(BigQuery.class);
  SinkTaskContext sinkTaskContext = mock(SinkTaskContext.class);
  InsertAllResponse insertAllResponse = mock(InsertAllResponse.class);

  when(bigQuery.insertAll(anyObject())).thenReturn(insertAllResponse);
  when(insertAllResponse.hasErrors()).thenReturn(false);

  BigQuerySinkTask testTask = new BigQuerySinkTask(bigQuery, null);
  testTask.initialize(sinkTaskContext);
  testTask.start(properties);

  testTask.put(Collections.singletonList(spoofSinkRecord(topic)));
  testTask.flush(Collections.emptyMap());
  verify(bigQuery, times(1)).insertAll(any(InsertAllRequest.class));
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
@Test
public void testSimplePutWhenSchemaRetrieverIsNotNull() {
  final String topic = "test-topic";

  Map<String, String> properties = propertiesFactory.getProperties();
  properties.put(BigQuerySinkConfig.TOPICS_CONFIG, topic);
  properties.put(BigQuerySinkConfig.DATASETS_CONFIG, ".*=scratch");

  BigQuery bigQuery = mock(BigQuery.class);
  SinkTaskContext sinkTaskContext = mock(SinkTaskContext.class);
  InsertAllResponse insertAllResponse = mock(InsertAllResponse.class);

  when(bigQuery.insertAll(anyObject())).thenReturn(insertAllResponse);
  when(insertAllResponse.hasErrors()).thenReturn(false);

  SchemaRetriever schemaRetriever = mock(SchemaRetriever.class);

  BigQuerySinkTask testTask = new BigQuerySinkTask(bigQuery, schemaRetriever);
  testTask.initialize(sinkTaskContext);
  testTask.start(properties);

  testTask.put(Collections.singletonList(spoofSinkRecord(topic)));
  testTask.flush(Collections.emptyMap());
  verify(bigQuery, times(1)).insertAll(any(InsertAllRequest.class));
  verify(schemaRetriever, times(1)).setLastSeenSchema(any(TableId.class), any(String.class), any(Schema.class));
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
@Test
public void testEmptyRecordPut() {
  final String topic = "test_topic";
  final Schema simpleSchema = SchemaBuilder
      .struct()
      .field("aField", Schema.STRING_SCHEMA)
      .build();

  Map<String, String> properties = propertiesFactory.getProperties();
  BigQuery bigQuery = mock(BigQuery.class);

  BigQuerySinkTask testTask = new BigQuerySinkTask(bigQuery, null);
  testTask.start(properties);

  SinkRecord emptyRecord = spoofSinkRecord(topic, simpleSchema, null);

  testTask.put(Collections.singletonList(emptyRecord));
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的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.BigQuery; //导入依赖的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.BigQuery; //导入依赖的package包/类
static public void loadSalesOrder(String jsonFileFullPath, BigQuery bigquery, BigQuerySnippets snippets, String datasetName, String tableName){
	try{
 	snippets.writeFileToTable(datasetName, tableName, getPath(jsonFileFullPath), FormatOptions.json());
 	System.out.println("A json.gz file was loaded into bigquery: " + jsonFileFullPath);
	}catch(IOException ioex){
		System.out.println(ioex.getMessage());
	}catch(InterruptedException interEx){
		
	}catch(TimeoutException toEx){
	}
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
private BigQuery getBigQuery() {
  if (testBigQuery != null) {
    return testBigQuery;
  }
  String projectName = config.getString(config.PROJECT_CONFIG);
  String keyFilename = config.getString(config.KEYFILE_CONFIG);
  return new BigQueryHelper().connect(projectName, keyFilename);
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
private BigQueryWriter getBigQueryWriter() {
  boolean updateSchemas = config.getBoolean(config.SCHEMA_UPDATE_CONFIG);
  int retry = config.getInt(config.BIGQUERY_RETRY_CONFIG);
  long retryWait = config.getLong(config.BIGQUERY_RETRY_WAIT_CONFIG);
  BigQuery bigQuery = getBigQuery();
  if (updateSchemas) {
    return new AdaptiveBigQueryWriter(bigQuery,
                                      getSchemaManager(bigQuery),
                                      retry,
                                      retryWait);
  } else {
    return new SimpleBigQueryWriter(bigQuery, retry, retryWait);
  }
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
private SchemaManager getSchemaManager(BigQuery bigQuery) {
  if (testSchemaManager != null) {
    return testSchemaManager;
  }
  SchemaRetriever schemaRetriever = config.getSchemaRetriever();
  SchemaConverter<com.google.cloud.bigquery.Schema> schemaConverter = config.getSchemaConverter();
  return new SchemaManager(schemaRetriever, schemaConverter, bigQuery);
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
private void ensureExistingTables(
    BigQuery bigQuery,
    SchemaManager schemaManager,
    Map<String, TableId> topicsToTableIds) {
  for (Map.Entry<String, TableId> topicToTableId : topicsToTableIds.entrySet()) {
    String topic = topicToTableId.getKey();
    TableId tableId = topicToTableId.getValue();
    if (bigQuery.getTable(tableId) == null) {
      logger.info("Table {} does not exist; attempting to create", tableId);
      schemaManager.createTable(tableId, topic);
    }
  }
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
/**
 * @param schemaRetriever Used to determine the Kafka Connect Schema that should be used for a
 *                        given table.
 * @param schemaConverter Used to convert Kafka Connect Schemas into BigQuery format.
 * @param bigQuery Used to communicate create/update requests to BigQuery.
 */
public SchemaManager(
    SchemaRetriever schemaRetriever,
    SchemaConverter<com.google.cloud.bigquery.Schema> schemaConverter,
    BigQuery bigQuery) {
  this.schemaRetriever = schemaRetriever;
  this.schemaConverter = schemaConverter;
  this.bigQuery = bigQuery;
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
/**
 * @param bigQuery Used to send write requests to BigQuery.
 * @param schemaManager Used to update BigQuery tables.
 * @param retry How many retries to make in the event of a 500/503 error.
 * @param retryWait How long to wait in between retries.
 */
public AdaptiveBigQueryWriter(BigQuery bigQuery,
                              SchemaManager schemaManager,
                              int retry,
                              long retryWait) {
  super(retry, retryWait);
  this.bigQuery = bigQuery;
  this.schemaManager = schemaManager;
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
/**
 * Clears tables in the given project and dataset, using a provided JSON service account key.
 */
public static void main(String[] args) {
  if (args.length < 4) {
    usage();
  }
  BigQuery bigQuery = new BigQueryHelper().connect(args[1], args[0]);
  for (int i = 3; i < args.length; i++) {
    if (bigQuery.delete(args[2], args[i])) {
      logger.info("Table {} in dataset {} deleted successfully", args[i], args[2]);
    } else {
      logger.info("Table {} in dataset {} does not exist", args[i], args[2]);
    }
  }
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
@Test
public void testBQTableDescription() {
  final String testTableName = "testTable";
  final String testDatasetName = "testDataset";
  final String testDoc = "test doc";
  final TableId tableId = TableId.of(testDatasetName, testTableName);

  SchemaRetriever mockSchemaRetriever = mock(SchemaRetriever.class);
  @SuppressWarnings("unchecked")
  SchemaConverter<com.google.cloud.bigquery.Schema> mockSchemaConverter =
      (SchemaConverter<com.google.cloud.bigquery.Schema>) mock(SchemaConverter.class);
  BigQuery mockBigQuery = mock(BigQuery.class);

  SchemaManager schemaManager = new SchemaManager(mockSchemaRetriever,
                                                  mockSchemaConverter,
                                                  mockBigQuery);

  Schema mockKafkaSchema = mock(Schema.class);
  // we would prefer to mock this class, but it is final.
  com.google.cloud.bigquery.Schema fakeBigQuerySchema =
      com.google.cloud.bigquery.Schema.of(Field.of("mock field", Field.Type.string()));

  when(mockSchemaConverter.convertSchema(mockKafkaSchema)).thenReturn(fakeBigQuerySchema);
  when(mockKafkaSchema.doc()).thenReturn(testDoc);

  TableInfo tableInfo = schemaManager.constructTableInfo(tableId, mockKafkaSchema);

  Assert.assertEquals("Kafka doc does not match BigQuery table description",
                      testDoc, tableInfo.getDescription());
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
@Test
public void testAutoCreateTables() {
  final String dataset = "scratch";
  final String existingTableTopic = "topic-with-existing-table";
  final String nonExistingTableTopic = "topic-without-existing-table";
  final TableId existingTable = TableId.of(dataset, "topic_with_existing_table");
  final TableId nonExistingTable = TableId.of(dataset, "topic_without_existing_table");

  Map<String, String> properties = propertiesFactory.getProperties();
  properties.put(BigQuerySinkConnectorConfig.TABLE_CREATE_CONFIG, "true");
  properties.put(BigQuerySinkConfig.SCHEMA_RETRIEVER_CONFIG, MockSchemaRetriever.class.getName());
  properties.put(BigQuerySinkConfig.SANITIZE_TOPICS_CONFIG, "true");
  properties.put(BigQuerySinkConfig.DATASETS_CONFIG, String.format(".*=%s", dataset));
  properties.put(
      BigQuerySinkConfig.TOPICS_CONFIG,
      String.format("%s, %s", existingTableTopic, nonExistingTableTopic)
  );

  BigQuery bigQuery = mock(BigQuery.class);
  Table fakeTable = mock(Table.class);
  when(bigQuery.getTable(existingTable)).thenReturn(fakeTable);
  when(bigQuery.getTable(nonExistingTable)).thenReturn(null);

  SchemaManager schemaManager = mock(SchemaManager.class);

  BigQuerySinkConnector testConnector = new BigQuerySinkConnector(bigQuery, schemaManager);
  testConnector.start(properties);

  verify(bigQuery).getTable(existingTable);
  verify(bigQuery).getTable(nonExistingTable);
  verify(schemaManager, never()).createTable(existingTable, existingTableTopic);
  verify(schemaManager).createTable(nonExistingTable, nonExistingTableTopic);
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
@Test
public void testEmptyPut() {
  Map<String, String> properties = propertiesFactory.getProperties();
  BigQuery bigQuery = mock(BigQuery.class);

  BigQuerySinkTask testTask = new BigQuerySinkTask(bigQuery, null);
  testTask.start(properties);

  testTask.put(Collections.emptyList());
}
 

import com.google.cloud.bigquery.BigQuery; //导入依赖的package包/类
@Test
public void testPutWhenPartitioningOnMessageTime() {
  final String topic = "test-topic";

  Map<String, String> properties = propertiesFactory.getProperties();
  properties.put(BigQuerySinkConfig.TOPICS_CONFIG, topic);
  properties.put(BigQuerySinkConfig.DATASETS_CONFIG, ".*=scratch");
  properties.put(BigQuerySinkTaskConfig.BIGQUERY_MESSAGE_TIME_PARTITIONING_CONFIG, "true");

  BigQuery bigQuery = mock(BigQuery.class);
  SinkTaskContext sinkTaskContext = mock(SinkTaskContext.class);
  InsertAllResponse insertAllResponse = mock(InsertAllResponse.class);

  when(bigQuery.insertAll(anyObject())).thenReturn(insertAllResponse);
  when(insertAllResponse.hasErrors()).thenReturn(false);

  BigQuerySinkTask testTask = new BigQuerySinkTask(bigQuery, null);
  testTask.initialize(sinkTaskContext);
  testTask.start(properties);

  testTask.put(Collections.singletonList(spoofSinkRecord(topic, "value", "message text", TimestampType.CREATE_TIME, 1509007584334L)));
  testTask.flush(Collections.emptyMap());
  ArgumentCaptor<InsertAllRequest> argument = ArgumentCaptor.forClass(InsertAllRequest.class);

  verify(bigQuery, times(1)).insertAll(argument.capture());
  assertEquals("test-topic$20171026", argument.getValue().getTable().getTable());
}