如何从 AppEngine 管道捕获 BigQuery 加载错误答案

【问题标题】：How to catch BigQuery loading errors from an AppEngine pipeline如何从 AppEngine 管道捕获 BigQuery 加载错误
【发布时间】：2014-06-14 07:24:15
【问题描述】：

我在 AppEngine 上构建了一个管道，将数据从 Cloud Storage 加载到 BigQuery。这工作正常，..直到有任何错误。如何通过 BigQuery 从我的 AppEngine 代码加载异常？

管道中的代码如下所示：

#Run the job
credentials = AppAssertionCredentials(scope=SCOPE)
http = credentials.authorize(httplib2.Http())
bigquery_service = build("bigquery", "v2", http=http)
jobCollection = bigquery_service.jobs()
result = jobCollection.insert(projectId=PROJECT_ID,
                              body=build_job_data(table_name, cloud_storage_files))
#Get the status
while (not allDone and not runtime.is_shutting_down()):
    try:
        job = jobCollection.get(projectId=PROJECT_ID,
                               jobId=insertResponse).execute()
        #Do something with job.get('status')
except:
    exc_type, exc_value, exc_traceback = sys.exc_info()
    logging.error(traceback.format_exception(exc_type, exc_value, exc_traceback))
    time.sleep(30)

这给了我状态错误或重大连接错误，但我正在寻找来自 BigQuery 的功能错误，例如字段格式转换错误、架构结构问题或 BigQuery 在尝试向表中插入行时可能遇到的其他问题。

如果 BigQuery 发生任何“功能性”错误，此代码将成功运行并正常完成，但不会在 BigQuery 上写入任何表。发生这种情况时不容易调试...

【问题讨论】：

标签： google-app-engine google-bigquery google-cloud-storage pipeline

【解决方案1】：

您可以使用异常中的 HTTP 错误代码。 BigQuery 是一个 REST API，因此返回的响应代码与 HTTP 错误代码 here 的描述相匹配。

这里有一些处理可重试错误（连接、速率限制等）的代码，但是当它是一个不期望的错误类型时会重新引发。

  except HttpError, err:
    # If the error is a rate limit or connection error, wait and
    # try again.
    # 403: Forbidden: Both access denied and rate limits.
    # 408: Timeout
    # 500: Internal Service Error
    # 503: Service Unavailable
    if err.resp.status in [403, 408, 500, 503]:
      print '%s: Retryable error %s, waiting' % (
          self.thread_id, err.resp.status,)
      time.sleep(5)
    else: raise

如果您想要更好的错误处理，请查看 bq 命令行客户端中的 BigqueryError 类（这曾经在 code.google.com 上可用，但随着最近切换到 gCloud，它不再可用。但如果你安装了 gcloud，bq.py 和 bigquery_client.py 文件应该在安装中）。

【讨论】：

事实证明，bq 命令行工具是获取 BigQuery 错误（如“提供的架构与表不匹配”）的唯一方法。但在从云存储加载数千个 csv 文件的工作中，要准确找出导致错误的文件非常困难。
bq 命令行工具只是 python 代码，它使用与您正在使用的相同的生成 python 客户端。这些错误可以从 bigquery 作业中读出。如果您使用检查 jobCollection.get(...).execute().get('status',[]).get('errorResult',{}) 您应该会看到类似“提供的架构不匹配”的错误

【解决方案2】：

这里的关键是粘贴的这部分代码：

except:
    exc_type, exc_value, exc_traceback = sys.exc_info()
    logging.error(traceback.format_exception(exc_type, exc_value, exc_traceback))
    time.sleep(30)

这个“例外”捕获每个异常，记录它，并让进程继续，而不考虑重试。

问题是，你想做什么呢？至少有“#Do something”评论的意图。

作为建议，请考虑使用 App Engine 的任务队列来检查状态，而不是等待 30 秒的循环。当任务出现异常时，它们会自动重试 - 您可以调整该行为。

【讨论】：