数据流作业失败并尝试在 Bigquery 上创建 temp_dataset

Posted 2023-03-25

【中文标题】数据流作业失败并尝试在 Bigquery 上创建 temp_dataset

【英文标题】：Dataflow job fails and tries to create temp_dataset on Bigquery

【发布时间】：2020-07-29 13:53:18

【问题描述】：

我正在运行一个简单的数据流作业来从一个表中读取数据并写回另一个表。 作业失败并出现错误：

工作流程失败。原因：S01：ReadFromBQ+WriteToBigQuery/WriteToBigQuery/NativeWrite 失败。，BigQuery 在项目“[我的项目]”中创建数据集“_dataflow_temp_dataset_18172136482196219053”失败。，BigQuery 执行失败。，错误： 消息：访问被拒绝：项目 [我的项目]：用户在项目 [我的项目] 中没有 bigquery.datasets.create 权限。

不过，我并没有尝试创建任何数据集，它基本上是在尝试创建一个 temp_dataset，因为作业失败了。但我没有得到任何关于幕后真正错误的信息。 阅读不是问题，真正失败的是写作步骤。我认为这与权限无关，但我的问题更多是关于如何获得真正的错误而不是这个错误。 知道如何处理这个问题吗？

代码如下：

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions, StandardOptions, WorkerOptions
from sys import argv

options = PipelineOptions(flags=argv)
google_cloud_options = options.view_as(GoogleCloudOptions)
google_cloud_options.project = "prj"
google_cloud_options.job_name = 'test'
google_cloud_options.service_account_email = "mysa"
google_cloud_options.staging_location = 'gs://'
google_cloud_options.temp_location = 'gs://'
options.view_as(StandardOptions).runner = 'DataflowRunner'
worker_options = options.view_as(WorkerOptions)
worker_options.subnetwork = 'subnet'

with beam.Pipeline(options=options) as p:
    query = "SELECT ..."

    bq_source = beam.io.BigQuerySource(query=query, use_standard_sql=True)

    bq_data = p | "ReadFromBQ" >> beam.io.Read(bq_source)

    table_schema = ...
    bq_data | beam.io.WriteToBigQuery(
        project="prj",
        dataset="test",
        table="test",
        schema=table_schema,
        create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
        write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
    )

【问题讨论】：

嗨，Alex，你能分享你的管道代码吗？另外，请写出 WriteToBigQuery 构造函数参数值，以便我们查看它是否来自配置错误。您能否在 IAM 中再次检查您是否拥有 bigquery.datasets.create 权限？

我应该说，请仔细检查 Dataflow 服务帐户是否具有该权限。

IIRC 创建临时数据集是 Dataflow 并行读取 BigQuery 机制的一部分。当您的服务帐号在您的项目中具有 BigQuery 写入权限时，您可以试试这是否可行？

所以添加 bigquery.datasets.create 权限解决了这个问题，但是试图授予最小权限，如果我想获得“真正的”错误，我真的没有办法给予这个权限

【参考方案1】：

使用 BigQuerySource 时，SDK 会创建一个临时数据集并将查询的输出存储到一个临时表中。然后它从该临时表发出导出以从中读取结果。

所以它创建这个 temp_dataset 是预期的行为。这意味着它可能没有隐藏错误。

这没有很好的文档记录，但可以通过读取调用在 BigQuerySource 的实现中看到：BigQuerySource.reader() --> BigQueryReader() --> BigQueryReader().__iter__() --> BigQueryWrapper.run_query() --> @ 987654325@.

【讨论】：

是的，我很难找到一个真正的解释，所以为什么我的问题！非常感谢您的快速回答和所有帮助！

您知道是否可以将现有数据集指定为临时数据集，而不是让 API 关联的服务帐户创建一个？