Google App Engine的BigQuery cron工作凭据

Question

更新：我已经缩小了问题的范围，所以我正在删除不必要的代码和示例：

更新2：在让cron作业以12小时的间隔运行相当长的一段时间后（每个都以成功告终，但没有用BQ编写任何内容），我们震惊地发现，大约一周后，其中一个cron作业确实成功写入BigQuery，以及声明“此请求导致为您的应用程序启动一个新进程（...）”的堆栈驱动程序日志，如下所述。以下作业再次停止写入。现在我想知道这是否与缓存的应用程序状态（有一些过期期限）或凭据过期日期有关，以某种方式阻止在第一次之后进一步写入BigQuery，但不会导致错误。

问题描述：

我正在尝试在App Engine（标准）中设置一个cron作业来查询并将数据写回BigQuery（数据集与部署的应用程序在同一项目中），cron作业成功执行，但仅在部署后的第一次执行时写入BigQuery，之后它们仍然成功执行但不写入。

我发现的主要区别在于StackDrive日志中，对于正确写入的执行，有额外的调试和信息，对于后续执行，没有这样的消息：

2018-04-19 04:44:03.933 CEST
Converted retries value: 3 -> Retry(total=3, connect=None, read=None, redirect=None, status=None) (/base/data/home/apps/e~<redacted>/lib/urllib3/util/retry.py:200)
2018-04-19 04:44:04.154 CEST
Making request: POST https://accounts.google.com/o/oauth2/token (/base/data/home/apps/e~<redacted>/lib/google/auth/transport/requests.py:117)
2018-04-19 04:44:04.160 CEST
Starting new HTTPS connection (1): accounts.google.com (/base/data/home/apps/e~<redacted>/lib/urllib3/connectionpool.py:824)
2018-04-19 04:44:04.329 CEST
https://accounts.google.com:443 "POST /o/oauth2/token HTTP/1.1" 200 None (/base/data/home/apps/e~<redacted>/lib/urllib3/connectionpool.py:396)
2018-04-19 04:44:04.339 CEST
Starting new HTTPS connection (1): www.googleapis.com (/base/data/home/apps/e~<redacted>/lib/urllib3/connectionpool.py:824)
2018-04-19 04:44:04.802 CEST
https://www.googleapis.com:443 "POST /bigquery/v2/projects/<redacted>/jobs HTTP/1.1" 200 None (/base/data/home/apps/e~<redacted>/lib/urllib3/connectionpool.py:396)
2018-04-19 04:44:04.813 CEST
This request caused a new process to be started for your application, and thus caused your application code to be loaded for the first time. This request may thus take longer and use more CPU than a typical request for your application.

我尝试过：

>

有人提到谷歌。标准应用程序引擎不完全支持云库，所以我尝试使用OAuth2/httplib2/GoogleAppClient凭据进行身份验证，但这是我第一次尝试，我不知道如果没有google，如何将各个部分组合在一起。云库我甚至不知道如何为BQ编写正确的查询

其他凭据设置方法如下所示，但连接到BQ似乎不是问题，它们都连接并写入（一次），只是在已部署的应用程序引擎中重复。

以下是完整的实施：

app.yaml：

runtime: python27
api_version: 1
threadsafe: true

handlers:
- url: /bigquerycron
  script: bigquerycron.app
  login: admin

libraries:
- name: ssl
  version: latest

env_variables:
  GAE_USE_SOCKETS_HTTPLIB : 'true'

bigquerycron。py公司

from __future__ import absolute_import
from google.cloud import bigquery
import webapp2

class MainPage(webapp2.RequestHandler):
    def get(self):
        self.response.headers['Content-Type'] = 'text/plain'
        self.response.write('CRON test page')          

def writeDataTest(dataset_id = '<redacted>',table_id='<redacted>'):
    client = bigquery.Client.from_service_account_json("credentials.json")
    job_config = bigquery.QueryJobConfig()
    table_ref = client.dataset(dataset_id).table(table_id)
    job_config.destination = table_ref
    job_config.write_disposition = 'WRITE_APPEND'

    query_job = client.query(
    """SELECT CURRENT_DATETIME() AS Datetime, 'CRON' as Source""", job_config=job_config)

writeDataTest()

app = webapp2.WSGIApplication([
      ('/bigquerycron', MainPage),
], debug=True)

克朗。yaml：

cron:
- url: /bigquerycron
  schedule: every 30 minutes

Answer 1

虽然我不确定原因，但我认为授权App引擎的服务号不足以访问BigQuery。

为了授权您的应用程序访问BigQuery，您可以使用以下两种方法之一：

>

在应用程序中。yaml文件，将指向服务帐户密钥文件的环境变量配置为BigQuery的正确授权配置：

env\u变量：GOOGLE\u APPLICATION\u凭据=[您的密钥文件]。json

您的代码执行从存储桶中获取授权服务号密钥，然后在Cloud Storage Client库的帮助下加载它。看到您的运行时是python，您应该使用的代码如下：

......

来自谷歌。云导入存储

......

def download\u key（）：

“”“从bucket下载密钥。”""

storage\u client=存储。客户端（）

bucket=storage\u客户端。get\u bucket（“YOURBUCKET”）

blob=铲斗。blob（'Keynameinthebucket.json'）

blob.download_to_filename（'Keynameinyourapp.json'）

......

#在代码中：

download_key（）

Client.from_service_account_jsonkeynameinyourapp.json

Answer 2

我怀疑如果您删除app.yaml的“login： admin”部分，它会起作用。

如果这是问题所在，请确保您有正确的X-Appengine标头设置

下面是一些用于任务队列和cron作业的文档。

Answer 3

在这种特定情况下，凭据不是问题，问题仅仅在于函数调用的位置，因为对应用程序引擎的工作方式有误解。bigquery的函数调用应该移动到主页类定义中，修复了bigquerycron。py如下所示（仅移动一行代码）：

from __future__ import absolute_import
from google.cloud import bigquery
import webapp2

class MainPage(webapp2.RequestHandler):
    def get(self):
        self.response.headers['Content-Type'] = 'text/plain'
        self.response.write('CRON test page')          
        writeDataTest()

def writeDataTest(dataset_id = '<redacted>',table_id='<redacted>'):
    client = bigquery.Client.from_service_account_json("credentials.json")
    job_config = bigquery.QueryJobConfig()
    table_ref = client.dataset(dataset_id).table(table_id)
    job_config.destination = table_ref
    job_config.write_disposition = 'WRITE_APPEND'
    query_job = client.query(
    """SELECT CURRENT_DATETIME() AS Datetime, 'CRON' as Source""", job_config=job_config)

app = webapp2.WSGIApplication([
      ('/bigquerycron', MainPage),
], debug=True)

OP中的版本实际上只向BigQuery写入一次，当第一次加载App Engine App时，所有后续调用都只执行MainPage类，在本例中，MainPage类什么都不做，因为实际的BigQuery代码在它之外。

此外，在不使用google cloud python库的情况下重写应用程序将是有益的，这在GAE标准中不受支持(https://github.com/GoogleCloudPlatform/google-cloud-python/issues/1893). 这尤其令人遗憾，因为即使是python的官方bigquery文档(https://cloud.google.com/bigquery/docs/)使用此库。然而，有多种变通方法可以继续使用它，包括链接的github问题中提到的一些方法，这里也提到了一些方法：在GAE中使用gcloud python，本例中使用了类似的变通方法。

但如前所述，最好使用专用于Python的Google API客户端库：https://developers.google.com/api-client-library/python/