import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
@Before
public void setup() throws Exception {
  when(bigqueryFactory.create(anyString(), anyString(), anyString())).thenReturn(bigquery);
  when(bigqueryFactory.create(
      anyString(),
      Matchers.any(HttpTransport.class),
      Matchers.any(JsonFactory.class),
      Matchers.any(HttpRequestInitializer.class)))
          .thenReturn(bigquery);

  when(bigquery.tabledata()).thenReturn(tabledata);
  when(tabledata.insertAll(
      anyString(),
      anyString(),
      anyString(),
      Matchers.any(TableDataInsertAllRequest.class))).thenReturn(insertAll);
  action = new MetricsExportAction();
  action.bigqueryFactory = bigqueryFactory;
  action.insertId = "insert id";
  action.parameters = parameters;
  action.projectId = "project id";
  action.tableId = "eppMetrics";
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
@Before
public void before() throws Exception {
  createTld("tld");

  action = new VerifyEntityIntegrityAction();
  action.mrRunner = new MapreduceRunner(Optional.of(2), Optional.of(2));
  action.response = new FakeResponse();
  BatchComponent component = mock(BatchComponent.class);
  inject.setStaticField(VerifyEntityIntegrityAction.class, "component", component);
  integrity =
      new VerifyEntityIntegrityStreamer(
          "project-id",
          bigqueryFactory,
          new Retrier(new FakeSleeper(new FakeClock()), 1),
          Suppliers.ofInstance("rowid"),
          now);
  when(bigqueryFactory.create(anyString(), anyString(), anyString())).thenReturn(bigquery);
  when(component.verifyEntityIntegrityStreamerFactory()).thenReturn(streamerFactory);
  when(streamerFactory.create(any(DateTime.class))).thenReturn(integrity);
  when(bigquery.tabledata()).thenReturn(bigqueryTableData);
  rowsCaptor = ArgumentCaptor.forClass(TableDataInsertAllRequest.class);
  when(bigqueryTableData.insertAll(anyString(), anyString(), anyString(), rowsCaptor.capture()))
          .thenReturn(bigqueryInsertAll);
  when(bigqueryInsertAll.execute()).thenReturn(new TableDataInsertAllResponse());

}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
@Test
public void test() throws IOException {
	String projectId = "";
	String datasetId = "ontologies";
	String tableId = "";
	String timestamp = Long.toString((new Date()).getTime());
	
	Bigquery bigquery = null;
	TableRow row = new TableRow();
	row.set("column_name", 7.7);
	TableDataInsertAllRequest.Rows rows = new TableDataInsertAllRequest.Rows();
	rows.setInsertId(timestamp);
	
	rows.setJson(row);
	List  rowList =
	    new ArrayList();
	rowList.add(rows);
	TableDataInsertAllRequest content = 
	    new TableDataInsertAllRequest().setRows(rowList);
	TableDataInsertAllResponse response =
	    bigquery.tabledata().insertAll(
	        projectId, datasetId, tableId, content).execute();
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
public void run() {
    try {
        // Prepare target table
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        Date now = new Date();
        String date = sdf.format(now);
        String targetTable = filterId + "_results_" + date + "_v" + parent.TABLE_STRUCTURE_VERSION;
        targetTable = targetTable.replace('-', '_');
        parent.prepareTable(targetTable);

        // Execute
        TableDataInsertAllRequest ir = new TableDataInsertAllRequest().setRows(rows);
        TableDataInsertAllResponse response = parent.bigquery.tabledata().insertAll(parent.projectId, parent.datasetId, targetTable, ir).execute();
        List<TableDataInsertAllResponse.InsertErrors> errors = response.getInsertErrors();
        if (errors != null) {
            LOG.error(errors.size() + " error(s) while writing " + filterId + " to BigQuery");
        } else {
            // Log lines for debug
            LOG.info(rows.size() + " lines written for " + filterId);
        }
    } catch (Exception e) {
        LOG.error("Failed to write to BigQuery", e);
    }
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
protected TableDataInsertAllRequest.Rows createRows(Event event) {
    TableDataInsertAllRequest.Rows insertRequestRows = null;
    try {
        insertRequestRows = insertRequestRowsBuilderFactory.createRows(event);
        if (logger.isDebugEnabled()) {
            if (insertRequestRows != null) {
                logger.debug("Row created from event '%s': %s", event, insertRequestRows.toPrettyString());
            } else {
                logger.debug("No row created from event '%s'", event);
            }
        }
    } catch (Exception ex) {
        logger.warn(String.format("Error creating rows from event '%s': %s. Skipping it.", event, ex.getMessage()), ex);
    }
    return insertRequestRows;
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
private void onInsertAll(List<List<Long>> errorIndicesSequence) throws Exception {
  when(mockClient.tabledata())
      .thenReturn(mockTabledata);

  final List<TableDataInsertAllResponse> responses = new ArrayList<>();
  for (List<Long> errorIndices : errorIndicesSequence) {
    List<TableDataInsertAllResponse.InsertErrors> errors = new ArrayList<>();
    for (long i : errorIndices) {
      TableDataInsertAllResponse.InsertErrors error =
          new TableDataInsertAllResponse.InsertErrors();
      error.setIndex(i);
    }
    TableDataInsertAllResponse response = new TableDataInsertAllResponse();
    response.setInsertErrors(errors);
    responses.add(response);
  }

  doAnswer(
      new Answer<Bigquery.Tabledata.InsertAll>() {
        @Override
        public Bigquery.Tabledata.InsertAll answer(InvocationOnMock invocation) throws Throwable {
          Bigquery.Tabledata.InsertAll mockInsertAll = mock(Bigquery.Tabledata.InsertAll.class);
          when(mockInsertAll.execute())
              .thenReturn(responses.get(0),
                  responses.subList(1, responses.size()).toArray(
                      new TableDataInsertAllResponse[responses.size() - 1]));
          return mockInsertAll;
        }
      })
      .when(mockTabledata)
      .insertAll(anyString(), anyString(), anyString(), any(TableDataInsertAllRequest.class));
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
/**
 * Check that the given conditional holds, and if not, stream a separate row to BigQuery for the
 * cross product of every supplied target and source (the message will be the same for each).
 * This is used in preference to records (repeated fields) in BigQuery because they are
 * significantly harder to work with.
 *
 * @return Whether the check succeeded.
 */
private <S, T> boolean checkManyToMany(
    boolean conditional,
    Iterable<S> sources,
    Iterable<T> targets,
    @Nullable String message) {
  if (conditional) {
    return true;
  }
  ImmutableList.Builder<Rows> rows = new ImmutableList.Builder<>();
  for (S source : sources) {
    for (T target : targets) {
      Map<String, Object> rowData =
          new ImmutableMap.Builder<String, Object>()
              .put(
                  FIELD_SCANTIME,
                  new com.google.api.client.util.DateTime(scanTime.toDate()))
              .put(
                  FIELD_SOURCE,
                  source.toString())
              .put(
                  FIELD_TARGET,
                  target.toString())
              .put(
                  FIELD_MESSAGE,
                  (message == null) ? NULL_STRING : message)
              .build();
      rows.add(
          new TableDataInsertAllRequest.Rows().setJson(rowData).setInsertId(idGenerator.get()));
    }
  }
  streamToBigqueryWithRetry(rows.build());
  return false;
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
/** Asserts that the given integrity errors, and no others, were logged to BigQuery. */
private void assertIntegrityErrors(IntegrityError... errors) {
  ImmutableList.Builder<Rows> expected = new ImmutableList.Builder<>();
  for (IntegrityError error : errors) {
    expected.add(new Rows().setInsertId("rowid").setJson(error.toMap(now)));
  }
  ImmutableList.Builder<Rows> allRows = new ImmutableList.Builder<>();
  for (TableDataInsertAllRequest req : rowsCaptor.getAllValues()) {
    allRows.addAll(req.getRows());
  }
  assertThat(allRows.build()).containsExactlyElementsIn(expected.build());
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
public static TableDataInsertAllResponse streamRow(Bigquery bigquery,
    String projectId,
    String datasetId,
    String tableId,
    TableDataInsertAllRequest.Rows row) throws IOException{
  
  return bigquery.tabledata().insertAll(
      projectId, 
      datasetId, 
      tableId, 
      new TableDataInsertAllRequest().setRows(Collections.singletonList(row))).execute();
  
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
@Before
public void setup() throws Exception {
  action.insertId = "insert id";
  action.path = "/path";
  action.method = "GET";
  action.tld = "tld";
  action.startTime = "0";
  action.endTime = "1";
  action.activity = "foo bar";
  action.responseCode = 200;
  action.projectId = "project id";
  action.bigquery = bigquery;
  when(bigquery.tabledata()).thenReturn(tabledata);
  when(tabledata.insertAll(
      "project id",
      "metrics",
      "streaming",
      new TableDataInsertAllRequest()
          .setRows(ImmutableList.of(new TableDataInsertAllRequest.Rows()
          .setInsertId("insert id")
          .setJson(new ImmutableMap.Builder<String, Object>()
              .put("path", "/path")
              .put("method", "GET")
              .put("tld", "tld")
              .put("start_time", "0")
              .put("end_time", "1")
              .put("response_code", 200)
              .put("activity", "foo bar")
              .build()))))).thenReturn(insertAll);
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
@Override
public TableDataInsertAllRequest.Rows createRows(Event event) {
    TableDataInsertAllRequest.Rows result = null;
    if (event != null) {
        TableRow row = new TableRow();
        Map<String, String> headers = new LinkedHashMap<String, String>(event.getHeaders());
        if (!headers.isEmpty()) {
            String id = idHeaderName != null ? headers.get(idHeaderName) : null;
            for (Map.Entry<String, String> headerEntry : headers.entrySet()) {
                String headerName = headerEntry.getKey();
                if (!excludeHeaders.isEmpty() && excludeHeaders.contains(headerName)) {
                    continue;
                }
                String columnName = includeHeadersRenameMap.get(headerName);
                if (!excludeHeaders.isEmpty() || columnName != null || includeHeadersRenameMap.isEmpty()) {
                    row = setColumnValue(row, columnName == null ? headerName : columnName, headerEntry.getValue());
                    if (row == null) {
                        break;
                    }
                }
            }
            if (row != null) {
                result = new TableDataInsertAllRequest.Rows();
                if (StringUtils.isNotBlank(id)) {
                    result.setInsertId(id);
                }
                result.setJson(row);
            }
        }
    }
    return result;
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
@Test
public void sendMessage() throws Exception {
    createProducer().process(createExchangeWithBody(new DefaultCamelContext(), new HashMap<>()));
    ArgumentCaptor<TableDataInsertAllRequest> dataCaptor = ArgumentCaptor.forClass(TableDataInsertAllRequest.class);
    Mockito.verify(tabledata).insertAll(Mockito.eq(TEST_PROJECT_ID), Mockito.eq(TEST_DATASET_ID), Mockito.eq(TEST_TABLE_ID), dataCaptor.capture());
    List<TableDataInsertAllRequest> requests = dataCaptor.getAllValues();
    Assert.assertEquals(1, requests.size());
    Assert.assertEquals(1, requests.get(0).getRows().size());
    Assert.assertNull(requests.get(0).getRows().get(0).getInsertId());
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
@Test
public void sendMessageWithTableId() throws Exception {
    Exchange exchange = createExchangeWithBody(new DefaultCamelContext(), new HashMap<>());
    exchange.getIn().setHeader(GoogleBigQueryConstants.TABLE_ID, "exchange_table_id");
    createProducer().process(exchange);
    ArgumentCaptor<TableDataInsertAllRequest> dataCaptor = ArgumentCaptor.forClass(TableDataInsertAllRequest.class);
    Mockito.verify(tabledata).insertAll(Mockito.eq(TEST_PROJECT_ID), Mockito.eq(TEST_DATASET_ID), Mockito.eq("exchange_table_id"), dataCaptor.capture());
    List<TableDataInsertAllRequest> requests = dataCaptor.getAllValues();
    Assert.assertEquals(1, requests.size());
    Assert.assertEquals(1, requests.get(0).getRows().size());
    Assert.assertNull(requests.get(0).getRows().get(0).getInsertId());
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
@Test
public void useAsInsertIdConfig() throws Exception {
    configuration.setUseAsInsertId("row1");
    Map<String, String> map = new HashMap<>();
    map.put("row1", "value1");
    createProducer().process(createExchangeWithBody(new DefaultCamelContext(), map));
    ArgumentCaptor<TableDataInsertAllRequest> dataCaptor = ArgumentCaptor.forClass(TableDataInsertAllRequest.class);
    Mockito.verify(tabledata).insertAll(Mockito.eq(TEST_PROJECT_ID), Mockito.eq(TEST_DATASET_ID), Mockito.eq(TEST_TABLE_ID), dataCaptor.capture());
    List<TableDataInsertAllRequest> requests = dataCaptor.getAllValues();
    Assert.assertEquals(1, requests.size());
    Assert.assertEquals(1, requests.get(0).getRows().size());
    Assert.assertEquals("value1", requests.get(0).getRows().get(0).getInsertId());
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
@Test
public void listOfMessages() throws Exception {
    List<Map<String, String>> messages = new ArrayList<>();
    messages.add(new HashMap<>());
    messages.add(new HashMap<>());
    createProducer().process(createExchangeWithBody(new DefaultCamelContext(), messages));
    ArgumentCaptor<TableDataInsertAllRequest> dataCaptor = ArgumentCaptor.forClass(TableDataInsertAllRequest.class);
    Mockito.verify(tabledata).insertAll(Mockito.eq(TEST_PROJECT_ID), Mockito.eq(TEST_DATASET_ID), Mockito.eq(TEST_TABLE_ID), dataCaptor.capture());
    List<TableDataInsertAllRequest> requests = dataCaptor.getAllValues();
    Assert.assertEquals(1, requests.size());
    Assert.assertEquals(2, requests.get(0).getRows().size());
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
private void verifyInsertAll(int expectedRetries) throws IOException {
  verify(mockClient, times(expectedRetries)).tabledata();
  verify(mockTabledata, times(expectedRetries))
      .insertAll(anyString(), anyString(), anyString(), any(TableDataInsertAllRequest.class));
}
 

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

    // Create source table
    String tableId = "data";
    String datasetId = BQ_TAG + "_extract_test";
    Dataset dataset = new Dataset().setDatasetReference(new DatasetReference()
            .setProjectId(gcpProjectId)
            .setDatasetId(datasetId));
    bq.datasets().insert(gcpProjectId, dataset)
            .execute();
    Table table = new Table().setTableReference(new TableReference()
            .setProjectId(gcpProjectId)
            .setTableId(tableId))
            .setSchema(new TableSchema()
                    .setFields(ImmutableList.of(
                            new TableFieldSchema().setName("foo").setType("STRING"),
                            new TableFieldSchema().setName("bar").setType("STRING")
                    )));
    bq.tables().insert(gcpProjectId, datasetId, table)
            .execute();

    // Populate source table
    TableDataInsertAllRequest content = new TableDataInsertAllRequest()
            .setRows(ImmutableList.of(
                    new TableDataInsertAllRequest.Rows().setJson(ImmutableMap.of(
                            "foo", "a",
                            "bar", "b")),
                    new TableDataInsertAllRequest.Rows().setJson(ImmutableMap.of(
                            "foo", "c",
                            "bar", "d"))));
    bq.tabledata().insertAll(gcpProjectId, datasetId, tableId, content)
            .execute();

    // Run extract
    String objectName = GCS_PREFIX + "test.csv";
    addWorkflow(projectDir, "acceptance/bigquery/extract.dig");
    Id attemptId = pushAndStart(server.endpoint(), projectDir, "extract", ImmutableMap.of(
            "src_dataset", datasetId,
            "src_table", tableId,
            "dst_bucket", GCS_TEST_BUCKET,
            "dst_object", objectName,
            "outfile", outfile.toString()));
    expect(Duration.ofMinutes(5), attemptSuccess(server.endpoint(), attemptId));
    assertThat(Files.exists(outfile), is(true));

    // Check that destination file was created
    StorageObject metadata = gcs.objects().get(GCS_TEST_BUCKET, objectName)
            .execute();
    assertThat(metadata.getName(), is(objectName));
    ByteArrayOutputStream data = new ByteArrayOutputStream();
    gcs.objects().get(GCS_TEST_BUCKET, objectName)
            .executeMediaAndDownloadTo(data);
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
public BigQueryInsertRunnable(BigQuerySinkBolt parent, String filterId, ArrayList<TableDataInsertAllRequest.Rows> rows) {
    this.parent = parent;
    this.filterId = filterId;
    this.rows = rows;
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllRequest; //导入依赖的package包/类
TableDataInsertAllRequest.Rows createRows(Event event);