无法将BigQuery表名作为ValueProvider传递给数据流模板

Question

我想把BigQuery表名作为运行时参数传递给我的数据流模板，就像这样简单：

class UserOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_value_provider_argument('--input', type=str, help='BigQuery table reference DATASET.TABLE')
        parser.add_value_provider_argument('--output', type=str, help='BigQuery table reference DATASET.TABLE')

def run(argv=None):    
    pipeline_options = PipelineOptions()
    p = beam.Pipeline(options=pipeline_options)
    user_options = pipeline_options.view_as(UserOptions)

    # Query
    query = f"""
    SELECT * FROM `{user_options.input}`
    WHERE last_scrape_date > (SELECT max(last_scrape_date) from `{user_options.output}`)
    """
    (p
     | 'Read from BQ Table' >> beam.io.Read(beam.io.BigQuerySource(query=query, use_standard_sql=True))
     | 'Write to BigQuery' >> beam.io.Write(
         beam.io.WriteToBigQuery(
             '{user_options.output}',
             schema=schema,
             # Creates the table in BigQuery if it does not yet exist.
             create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
             # Deletes all data in the BigQuery table before writing.
             write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE)))
    p.run().wait_until_finish()

apitools.base.py.exceptions.HttpBadRequestError: HttpError accessing <https://bigquery.googleapis.com/bigquery/v2/projects/vf-scrapers/jobs?alt=json>: response: <{'vary': 'Origin, X-Origin, Referer', 'content-type': 'application/json; charset=UTF-8', 'date': 'Tue, 15 Dec 2020 23:25:17 GMT', 'server': 'ESF', 'cache-control': 'private', 'x-xss-protection': '0', 'x-frame-options': 'SAMEORIGIN', 'x-content-type-options': 'nosniff', 'transfer-encoding': 'chunked', 'status': '400', 'content-length': '500', '-content-encoding': 'gzip'}>, content <{
  "error": {
    "code": 400,
    "message": "Table name \"RuntimeValueProvider(option: input, type: str, default_value: None)\" missing dataset while no default dataset is set in the request.",
    "errors": [
      {
        "message": "Table name \"RuntimeValueProvider(option: input, type: str, default_value: None)\" missing dataset while no default dataset is set in the request.",
        "domain": "global",
        "reason": "invalid"
      }
    ],
    "status": "INVALID_ARGUMENT"
  }
}

这个问题有一个解决方案:https://issues.apache.org/jira/browse/beam-1440，但到目前为止我还不明白结论。

Answer 1

运行时参数，如您的Bigquery表名，在管道构建过程中表示为valueProvider对象，而不是字符串文本。您可以在输出中看到user_options.input作为字符串化的runtimeValueProvider输出。但在这种情况下，修复非常简单；直接传入对象，而不将其转换为字符串，如下所示：

beam.io.WriteToBigQuery(
             user_options.output,
             schema=schema,
             ...

这是因为在数据流模板中，管道构造发生在传递运行时参数之前。调用此代码时，您的输入和输出参数尚未定义。相反，ValueProvider充当占位符，一旦定义了参数，就允许在运行时检索参数。