Google Cloud Dataflow - WriteToBigQuery：“NoneType”对象没有属性“__getitem__”答案

【问题标题】：Google Cloud Dataflow - WriteToBigQuery: 'NoneType' object has no attribute '__getitem__'Google Cloud Dataflow - WriteToBigQuery：“NoneType”对象没有属性“__getitem__”
【发布时间】：2019-05-01 13:40:53
【问题描述】：

我在尝试运行 GCP Cloud-Dataflow 管道时遇到问题。

管道在使用“DirectRunner”在本地运行时有效，但在尝试使用“DataflowRunner”在数据流中运行时失败。

当在管道上调用run() 并显示上面给出的错误消息时，它立即失败（而不是先部署到 GCP，然后在实际运行管道时失败）。

异常是在对beam.io.WriteToBigQuery的调用中引发的：

(bq_rows 
| 'map_to_row' >> beam.Map(to_pred_row)
| 'write_to_table' >> beam.io.WriteToBigQuery(
    'my_dataset_name.my_table_name', 
    write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
    create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED))

如果我将管道中的最后一个节点替换为仅写入文件的内容：

(bq_rows 
| 'map_to_row' >> beam.Map(to_pred_row)
| 'debug_write_to_csv_2' >> beam.io.WriteToText(additional_opts.out_path, ".txt"))

然后一切都按预期工作，我得到一个包含我期望的所有记录的文本文件。

如果我使用 WriteToBigQuery() 函数按原样运行所有内容，但改回 DirectRunner（不做任何更改），那么一切正常，新行将写入 BQ 表。

据我所知，流入的记录并没有什么特别之处 WriteToBigQuery 节点。我已将这些输出到本地和云中运行的文本文件，以找出导致此错误的原因，但两个输出看起来相同（并且与目标表的架构匹配）。无论如何，在运行流程时，对于意外的值或参数，事情似乎还远远不够——正如我提到的，每当我在管道上调用 run() 时就会发生此错误

我哪里错了？

更新：

这是相同行为的最小示例。在创建了一个名为 temp_e.words 的表和一个名为 word 的单个（STRING，REQUIRED）列之后，我可以使用以下代码重现该行为：

import apache_beam as beam
from google.cloud import storage as gcs
import shutil
from google.cloud import bigquery as bq
import datetime
import os
import json
import apache_beam as beam
from apache_beam.options.pipeline_options import (GoogleCloudOptions, 
                                                  StandardOptions)


def to_row(word):
  return {
      'word': word
  }

def run_pipeline(local_mode):

  PROJECT = 'searchlab-data-insights'
  REGION = 'us-central1'
  GCS_BUCKET_PATH = 'gs://temp-staging-e'

  timestamp = datetime.datetime.now().strftime('%y%m%d-%H%M%S')

  options = beam.pipeline.PipelineOptions([
      '--project', PROJECT
  ])

  if local_mode:
    RUNNER = 'DirectRunner'
  else:
    RUNNER = 'DataflowRunner'

  google_cloud_options = options.view_as(GoogleCloudOptions)
  google_cloud_options.project = PROJECT
  google_cloud_options.job_name = 'test-{}'.format(timestamp)
  google_cloud_options.staging_location = os.path.join(GCS_BUCKET_PATH, 'staging')
  google_cloud_options.temp_location = os.path.join(GCS_BUCKET_PATH, 'tmp')
  options.view_as(StandardOptions).runner = RUNNER

  p = beam.Pipeline(RUNNER, options=options)
  bq_rows = p | beam.Create(['words', 'to', 'store']) 

  (bq_rows 
    | 'map_to_row' >> beam.Map(to_row)
    | 'write_to_table' >> beam.io.WriteToBigQuery(
        'temp_e.words', 
        write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
        create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)
  )

  job = p.run()

  if local_mode:
    job.wait_until_finish()
    print "Done!"

现在运行 run_pipeline(local_mode=True) 会产生正确的结果并附加行，而运行 run_pipeline(local_mode=False) 会立即触发错误。

生成的完整错误在这里：https://pastebin.com/xx8wwtXV

【问题讨论】：

显示您的代码。你的问题是write_to_predictions_table 的问题。什么是 Stackdriver 报告？
您能否添加更多有关您的错误的信息，例如堆栈跟踪或行号？还要确保输入数据集中没有空行。
嗨，我已经用一个可重现的例子更新了这个问题。抱歉，这不是第一次包含在内。创建此示例后，我可以看到该行为与数据流运行器读取预先存在的模式的能力有关。如果我更改为新表，本地运行器也会失败，并显示一条有意义的错误消息，要求我提供模式。显式提供模式（即使表已经存在）是解决我最初问题的一种解决方法。感谢您的帮助。
@JohnHanley 感谢您的帮助 - 没有向 stackdriver 写入任何内容，因为该过程并未真正在 GCP 上启动该作业。调用.run()时失败

标签： python google-cloud-platform google-bigquery google-cloud-dataflow apache-beam

【解决方案1】：

只有在没有为调用 beam.io.WriteToBigQuery 提供架构时才会出现此问题。似乎 DirectRunner 可以使用现有的表模式工作，但 DataflowRunner 不能。

如果没有更好的答案，我们可以通过显式提供架构来解决它。

因此，例如，在上面的最小示例中，我们可以使用：

(bq_rows | 'map_to__row' >> beam.Map(to_row) | 'write_to_table' >> beam.io.WriteToBigQuery( 'temp_e.words', schema={"fields":[{"type":"STRING","name":"word","mode":"REQUIRED"}]} write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND, create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED) )

【讨论】：