如何在 Java 中正确迭代 Bigquery TableResult答案

【问题标题】：How to properly iterate Bigquery TableResult in Java如何在 Java 中正确迭代 Bigquery TableResult
【发布时间】：2023-02-09 18:11:18
【问题描述】：

我正在尝试使用 getValues() 迭代 TableResult 中的行，如下所示。如果我使用getValues()，它只会检索第一页的行。我想使用getValues()而不是iterateAll()来迭代所有行。在下面的代码中，问题是它的时间是无限的。没有结束。 while(results.hasNextPage())没有结束。下面的代码有什么问题？

    {
    query = "select from aa.bb.cc";
    QueryJobConfiguration queryConfig =
            QueryJobConfiguration.newBuilder(query)
                    .setPriority(QueryJobConfiguration.Priority.BATCH)
                    .build();
    TableResult results = bigquery.query(queryConfig);

    int i = 0;
    int j=0;
    while(results.hasNextPage()) {
        j++;
        System.out.println("page " + j);
        System.out.println("Data Extracted::" + i + " records");
        for (FieldValueList row : results.getNextPage().getValues()) {
            i++;
        }
    }
    System.out.println("Total Count::" + results.getTotalRows());
    System.out.println("Data Extracted::" + i + " records");
}

我在源表中只有 200,000 条记录。下面是输出，我强行停止了这个过程。

page 1
Data Extracted::0 records
page 2
Data Extracted::85242 records
page 3
Data Extracted::170484 records
page 4
Data Extracted::255726 records
page 5
Data Extracted::340968 records
page 6
Data Extracted::426210 records
page 7
Data Extracted::511452 records
page 8
Data Extracted::596694 records
.......
.......
.......
.......

【问题讨论】：

标签： java google-cloud-platform google-bigquery gcs

【解决方案1】：

简而言之，您需要使用 getNextPage() 变量更新 TableResults 变量。如果您不更新它，您将始终一遍又一遍地循环相同的结果。这就是为什么您在输出中获得大量记录的原因。

如果您检查以下示例：Bigquery Pagination 和Using Java Client Library。我们可以通过多种方式处理分页结果。虽然不是特定于单次运行查询。

如下面的代码所示，它部分基于分页示例，您需要使用 getNextPage() 的输出来更新 results 变量并继续在 while up 内执行下一次迭代，直到迭代所有页面，但最后一次.

QueryRun.Java

package com.projects;

// [START bigquery_query]
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryException;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.QueryJobConfiguration; 
import com.google.cloud.bigquery.TableResult;
import com.google.cloud.bigquery.Job;
import com.google.cloud.bigquery.JobId;
import com.google.cloud.bigquery.FieldValueList;
import com.google.cloud.bigquery.JobInfo;
import com.google.cloud.bigquery.BigQuery.QueryResultsOption;
import java.util.UUID;

import sun.jvm.hotspot.debugger.Page;

public class QueryRun {

  public static void main(String[] args) {

    String projectId = "bigquery-public-data";
    String datasetName = "covid19_ecdc_eu";
    String tableName = "covid_19_geographic_distribution_worldwide";
    String query =
        "SELECT * "
            + " FROM `"
            + projectId
            + "."
            + datasetName
            + "."
            + tableName
            + "`"
            + " LIMIT 100";
    System.out.println(query);
    query(query);
  }

  public static void query(String query) {
    try {
      BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
      QueryJobConfiguration queryConfig = QueryJobConfiguration.newBuilder(query).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());

      TableResult results = queryJob.getQueryResults(QueryResultsOption.pageSize(10));

      int i = 0;
      int j =0; 

      // get all paged data except last line
      while(results.hasNextPage()) {
        j++;   
        for (FieldValueList row : results.getValues()) { 
            i++;
        }
        results = results.getNextPage();
        print_msg(i,j);
      }

      // last line run
      j++;
      for (FieldValueList row : results.getValues()) {  
        i++;
      }
      print_msg(i,j);

      System.out.println("Query performed successfully.");
    } catch (BigQueryException | InterruptedException e) {
      System.out.println("Query not performed 
" + e.toString());
    }
  }

  public static void print_msg(int i,int j)
  {
    System.out.println("page " + j);
    System.out.println("Data Extracted::" + i + " records"); 
  }

} 
// [END bigquery_query]

输出：

SELECT *  FROM `bigquery-public-data.covid19_ecdc_eu.covid_19_geographic_distribution_worldwide` LIMIT 100
page 1
Data Extracted::10 records
page 2
Data Extracted::20 records
page 3
Data Extracted::30 records
page 4
Data Extracted::40 records
page 5
Data Extracted::50 records
page 6
Data Extracted::60 records
page 7
Data Extracted::70 records
page 8
Data Extracted::80 records
page 9
Data Extracted::90 records
page 10
Data Extracted::100 records
Query performed successfully.

最后一点，没有关于查询分页的官方示例，所以我不完全确定用 java 处理分页的推荐方法。 BigQuery for Java 文档页面上不是很清楚。如果您可以使用分页方法更新您的问题，我将不胜感激。

如果您在运行附加示例时遇到问题，请参阅Using the BigQuery Java client示例、其github页面及其内部的pom.xml文件，并检查您是否遵守它。

【讨论】：

使用 results.getValues() 而不是分页和 results.iterateAll() 时，性能会有所提高吗？
应该有 getValues 和分页它比 iterateAll 更好，它带来了所有记录。在巨大的结果上应该注意差异。另外，我没有检查是否有办法与此异步读取结果。
感谢你的回复。真的有帮助。还有一个问题，您能否分享一下关于处理 rateLimitExceeded 错误的想法。不是 Spring Boot 应用程序。我们如何最好地处理这个问题？
我认为您将不得不提出一个新问题来指定触发此类消息的场景的详细信息，因为需要更多详细信息才能提供准确的答案。猜测一下，它会不会是this之类的东西？如果是这样，您将必须检查您的服务限制和设置，也许您可以增加或管理此类限制/约束。

【解决方案2】：

可能是我回复晚了。但是阅读 java 客户端指南 (https://cloud.google.com/bigquery/docs/quickstarts/quickstart-client-libraries#complete_source_code)

它说：

迭代 QueryResponse 以获取结果中的所有行。迭代器自动处理分页。每个 FieldList 按数字索引或列名公开列。

也就是说，简单地使用 iterateAll() 方法应该更容易。

如果我错了，请告诉我。

【讨论】：