使用 bigquery 时增加 tidyverse API OAuth 令牌过期之前的时间？

Posted 2023-03-24

【中文标题】使用 bigquery 时增加 tidyverse API OAuth 令牌过期之前的时间？

【英文标题】：Increase time before tidyverse API OAuth token expires when using bigquery?

【发布时间】：2020-05-08 10:06:29

【问题描述】：

当using bigrquery from R 时，tidyverse API 要求进行身份验证，一旦提供身份验证，就可以从 R 会话访问 bigquery。下载结果可能很耗时，如果 oauth 令牌在传输过程中过期，则下载失败：

错误：无效凭据 [authError]

示例代码

# install.packages("tidyverse"); install.packages('bigrquery')
Sys.setenv(BIGQUERY_TEST_PROJECT="your-gcp-project-id") 
library(tidyverse)
library(dbplyr)
library(bigrquery)

billing <- bq_test_project()

connection <- dbConnect(
  bigrquery::bigquery(),
  project = "your-gcp-project-id",
  dataset = "dataset-name",
  billing = billing
)

mytable <- tbl(connection, "mytable")

mytable %>%
  -- some heavy dplyr wrangling --
  %>% collect()

返回

Running job [/] 20s
Complete
Billed 400GB
Downloading 230,000,000 rows
Downloading data [=====--------] 28% ETA: 24h

但过了一段时间

Error: Invalid Credentials [authError]

问题

如何增加 0Auth 令牌过期前的时间长度？

【问题讨论】：

下载 10GB 的时间与消费者的互联网速度相比似乎非常慢。您能否确认您的 R 内存足够大，可以将整个 10GB 表放入内存？否则，R 要么由于内存不足而挂起，要么正在写入和写入磁盘，这会导致额外的延迟。

@Simon.S.A.我同意速度，它很慢，但我已经尝试了很多次，结果相同：bigquery 运行速度非常快，但下载结果数据集很慢。我可以确认内存没有问题（30gb 实例没有任何其他运行，数据提取最大为 15gb）。我想知道它是否正在写入磁盘。我知道 a) 数据在 bigquery 中以 parquet 形式开始，由 bigquery 查询，然后结果神奇地出现在 R 会话中。简短的回答：我不知道他们是如何到达那里的（如果是直接写入内存还是先写入磁盘）

DBI 和 bigrquery 文档建议在不支持通过 dbplyr 连接到 bigquery 时设置超时参数。 bigquery 端可能有一个设置（即不是通过 R），但这超出了我的经验。

如果您的查询/表中有唯一的 ID 号，那么我可以提供我使用的解决方法。

@Simon.S.A.我想我知道您要建议什么，将查询拆分为多个较短的查询并在每个查询之间进行身份验证以避免令牌过期问题？我不得不这样做作为一种 hacky 解决方法，但每次都必须这样做会很糟糕

【参考方案1】：

我也有同样的错误，我得到了问题的超时。

另一个潜在的解决方案是导出到谷歌数据工作室，然后通过 csv 下载它作为一种解决方法。

或者对于大型数据集，最好通过将 BigQuery 结果提取到 Google Cloud Storage，然后从那里下载数据来完成，如本文所述： https://mran.microsoft.com/web/packages/bigQueryR/vignettes/bigQueryR.html

    ## Create the data extract from BigQuery to Cloud Storage
    job_extract <- bqr_extract_data("your_project",
                                    "your_dataset",
                                    "bigResultTable",
                                    "your_cloud_storage_bucket_name")

    ## poll the extract job to check its status
    ## its done when job$status$state == "DONE"
    bqr_get_job("your_project", job_extract$jobReference$jobId)

    ## to download via a URL and not logging in via Google Cloud Storage interface:
    ## Use an email that is Google account enabled
    ## Requires scopes:
    ##  https://www.googleapis.com/auth/devstorage.full_control
    ##  https://www.googleapis.com/auth/cloud-platform
    ## set via options("bigQueryR.scopes") and reauthenticate if needed

    download_url <- bqr_grant_extract_access(job_extract, "[e-mail]")

    ## download_url may be multiple if the data is > 1GB
    > [1] "https://storage.cloud.google.com/big-query-r-extracts/extract-20160311112410-000000000000.csv"
    > [2] "https://storage.cloud.google.com/big-query-r-extracts/extract-20160311112410-000000000001.csv"
    > [3] "https://storage.cloud.google.com/big-query-r-extracts/extract-20160311112410-000000000002.csv"

【参考方案2】：

不是一个完整的答案，而是我迄今为止的研究细节，以防止人们追溯我的步骤。

似乎无法通过 dbplyr 控制超时

dbconnect 接受驱动程序和要传递给驱动程序的参数 (documentation)：dbConnect(drv, ...)。 对于某些连接类型，附加参数可以包括timeout。 This issue 有一个使用 Cassandra 的示例：con <- dbConnect(odbc::odbc(), "Cassandra (DSN)", timeout = 10)。 但似乎不支持将超时作为bigquery 的参数。 documentation 列出了以下参数（project、dataset、billing、page_size、quiet、use_legacy_sql、bigint）并指出其他参数当前被忽略。

因此，鉴于上述情况，R 似乎不太可能通过dbplyr、DBI 或连接来控制超时。

将查询拆分为多个较短的查询

虽然不是 OP 的偏好（cmets 明确表示），但这仍然是一个潜在的解决方案。我使用一种基于唯一 ID 列的过滤方法，并使用包装函数来减少一些额外的混乱：

reconnect <- function(jj)
  if(exists("connection"))
    dbDisconnect(connection) # avoids multiple concurrent connections

  print(paste0(as.character(Sys.time()),"-- modulo ",jj," begun")) # track progress

  connection <- dbConnect(
    bigrquery::bigquery(),
    project = "your-gcp-project-id",
    dataset = "dataset-name",
    billing = billing
  )

  mytable <- tbl(connection, "mytable") %>%
    filter(unique_id %% NUM_SUBSETS == jj) # filter to subset, requires unique_id

  # assignment into the parent environment
  assign("connection", connection, envir = parent.frame())
  assign("mytable ", mytable , envir = parent.frame())

然后我们如下迭代：

## parameters
DEVELOPMENT_MODE = FALSE
NUM_SUBSETS = 50

## subset
modulo = if(DEVELOPMENT_MODE) modulo = 0 # only one if development mode
 else  modulo = 0:(NUM_SUBSETS-1) # otherwise all of them


results = data.frame()

for(jj in modulo)
  reconnect(jj)

  these_results = mytable %>%
    -- some heavy dplyr wrangling --
    %>% collect()

  results = rbind(results, these_results)

我在测试/开发时将 DEVELOPER_MODE 设置为 true，当我希望整个程序运行时设置为 false。

其他可以考虑的途径

检查是否可以在 bigquery 帐户中设置/控制超时（如果无法通过 R 控制）。 调查-- heavy dplyr wrangling here -- 的复杂程度。因为 dbplyr 不能翻译非常高效的 sql 代码，所以在我在 SQL 服务器上的工作中，保存中间表已经缩短了我的运行时间。鉴于下载 10GB 应该比几个小时快得多，瓶颈可能是 bigquery 执行所有动态处理（并且最初的 20 秒执行是延迟评估）。 This 链接表明单次执行的持续时间有 6 小时限制。