BigQuery 性能和批处理请求中的运行作业

Question

我们正在使用 Google BigQuery 处理 Java Web 应用程序，但遇到了一个奇怪的行为。

我们使用查询作业来检索数据，然后在一些图表中将它们可视化。

我们创建并添加多个职位：

Job query1Job = startAsyncQuery(query1, "q1"+uuid);
jobMapping.put(query1Job.getJobReference().getJobId(), "q1");
runningJobs.add(query1Job);
...
Job query2Job = startAsyncQuery(query1, "q2"+uuid);
jobMapping.put(query2Job.getJobReference().getJobId(), "q2");
runningJobs.add(query2Job);
...
Job query3Job = startAsyncQuery(query1, "q3"+uuid);
jobMapping.put(query1Job.getJobReference().getJobId(), "q3");
runningJobs.add(query3Job);
...
Job query4Job = startAsyncQuery(query1, "q4"+uuid);
jobMapping.put(query4Job.getJobReference().getJobId(), "q4");
runningJobs.add(query4Job);
...
Job query5Job = startAsyncQuery(query1, "q5"+uuid);
jobMapping.put(query1Job.getJobReference().getJobId(), "q5");
runningJobs.add(query5Job);


 public Job startAsyncQuery(String query, String jobId) throws IOException {
        JobConfigurationQuery queryConfig = new JobConfigurationQuery().setQuery(query).setUseQueryCache(true);
        JobConfiguration config = new JobConfiguration().setQuery(queryConfig);
        Job job = new Job().setId(jobId).setConfiguration(config);
        Job queuedJob = this.bigquery.jobs().insert(this.projectId, job).execute();
        return queuedJob;
    }

我们轮询正在运行的作业列表以检索数据：

  boolean isError = false;
        while (!runningJobs.isEmpty() && !isError) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            List<Job> tempJobs = new ArrayList<Job>();
            for (Job job : runningJobs) {
                JobReference jref = job.getJobReference();
                String jid = jobMapping.get(jref.getJobId());
                int jobState = pollJob(bigQueryManager, jref, jid);
                if (jobState == -1) {
                    System.out.println("Aborting because of error for job " + jid);
                    isError = true;
                } else if (jobState == 1) {
                    List<TableRow> rows = bigQueryManager.getQueryResults(jref);
                    if (jid.startsWith("q1")) {
                        parseQ1QueryResult(filter, metrics, metricsRTLDiv, clientList, objectList, rows);
                    } else if (jid.startsWith("q2")) {
                        parseQ2QueryResult(filter, metrics, rows);
                    } else if (jid.startsWith("q3")) {
                        parseQ3QueryResult(filter, metrics, metricsRTLDiv, rows);
                    } else if (jid.startsWith("q4")) {
                        parseQ4QueryResult(metricsRTLDiv, rows);
                    } else if (jid.startsWith("q5")) {
                        parseQ5QueryResult(metrics, rows);
                    } else {
                        System.out.println("Job finished for unknown id: " + jid);
                    }
                } else {
                    tempJobs.add(job);
                }               
            }
            runningJobs = tempJobs;
        }

奇怪的行为是 bigquery.jobs().insert.execute() 每次都需要几秒钟：

Fri Mar 20 08:57:56 CET 2015 - q1
Fri Mar 20 08:57:59 CET 2015 - q2
Fri Mar 20 08:58:04 CET 2015 - q3
Fri Mar 20 08:58:09 CET 2015 - q4
Fri Mar 20 08:58:14 CET 2015 - q5 

是否可以将所有这些查询放在一个批处理请求中，这将是一个 HTTP 请求（而不是每个查询一个 http 请求）？

当我们需要执行多个查询时，有人知道从 BigQuery 表中检索数据的快速方法吗？ 有没有办法提高作业执行速度？

谢谢。

Answer 1

适用于 Java 的 Google API 客户端库支持 Request Batching。

虽然此示例适用于日历服务，但它可以适应 BigQuery。

JsonBatchCallback<Calendar> callback = new JsonBatchCallback<Calendar>() {

  public void onSuccess(Calendar calendar, HttpHeaders responseHeaders) {
    printCalendar(calendar);
    addedCalendarsUsingBatch.add(calendar);
  }

  public void onFailure(GoogleJsonError e, HttpHeaders responseHeaders) {
    System.out.println("Error Message: " + e.getMessage());
  }
};

...

Calendar client = Calendar.builder(transport, jsonFactory, credential)
  .setApplicationName("BatchExample/1.0").build();
BatchRequest batch = client.batch();

Calendar entry1 = new Calendar().setSummary("Calendar for Testing 1");
client.calendars().insert(entry1).queue(batch, callback);

Calendar entry2 = new Calendar().setSummary("Calendar for Testing 2");
client.calendars().insert(entry2).queue(batch, callback);

batch.execute();