更改谷歌云数据流BigQuery优先级

Question

我有一个运行在从BigQuery读取数据的谷歌云DataFlow上的Beam作业。当我运行作业时，作业需要几分钟时间才能开始从(微型)表中读取数据。结果是数据流作业发送的是以批处理模式运行而不是以交互模式运行的BigQuery作业。如何将其切换为在Apache光束中立即运行？我在API中找不到更改优先级的方法。

Answer 1

也许Googler会纠正我，但是不能，你不能把它从BATCH改成INTERACTIVE，因为它不是由Beam公开的。

来自org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.java (here)：

private void executeQuery(
    String executingProject,
    String jobId,
    TableReference destinationTable,
    JobService jobService) throws IOException, InterruptedException {
  JobReference jobRef = new JobReference()
      .setProjectId(executingProject)
      .setJobId(jobId);

  JobConfigurationQuery queryConfig = createBasicQueryConfig()
      .setAllowLargeResults(true)
      .setCreateDisposition("CREATE_IF_NEEDED")
      .setDestinationTable(destinationTable)
      .setPriority("BATCH") <-- NOT EXPOSED
      .setWriteDisposition("WRITE_EMPTY");

  jobService.startQueryJob(jobRef, queryConfig);
  Job job = jobService.pollJob(jobRef, JOB_POLL_MAX_RETRIES);
  if (parseStatus(job) != Status.SUCCEEDED) {
    throw new IOException(String.format(
        "Query job %s failed, status: %s.", jobId, statusToPrettyString(job.getStatus())));
  }
}

如果查询在BATCH模式下运行对您来说确实是一个问题，那么一种解决方法可能是：

使用Beam API直接读取临时表，滚动您自己的初始请求，并设置优先级以将步骤1的结果INTERACTIVE.

• Write到BigQuery管道中的临时表

• ，使用BigQueryIO.Read.from()

读取临时表

Answer 2

您可以通过传递优先级参数来配置以“交互”优先级运行查询。有关详细信息，请查看此Github example。

请注意，您可能会到达其中一个BigQuery limits and quotas，因为当您使用批处理时，如果您曾经达到速率限制，查询将排队并稍后重试。与交互式限制相反，当达到这些限制时，查询将立即失败。这是因为BigQuery假设您需要立即运行交互式查询。