修改 MinimalWordCount 示例以从 BigQuery 中读取

Question

我正在尝试修改 Apache Beam 的 MinimalWordCount python 示例以从 BigQuery 表中读取。我已经进行了以下修改，并且我似乎使查询正常工作，但示例。

这里的原始示例：

 with beam.Pipeline(options=pipeline_options) as p:

    # Read the text file[pattern] into a PCollection.
    lines = p | ReadFromText(known_args.input)

    # Count the occurrences of each word.
    counts = (
        lines
        | 'Split' >> (beam.FlatMap(lambda x: re.findall(r'[A-Za-z\']+', x))
                      .with_output_types(unicode))
        | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
        | 'GroupAndSum' >> beam.CombinePerKey(sum))

    # Format the counts into a PCollection of strings.
    output = counts | 'Format' >> beam.Map(lambda (w, c): '%s: %s' % (w, c))

    # Write the output using a "Write" transform that has side effects.
    # pylint: disable=expression-not-assigned
    output | WriteToText(known_args.output)

而不是 ReadFromText 我正在尝试调整它以从 BigQuery 表中的列中读取。为此，我已将 lines = p | ReadFromText(known_args.input) 替换为以下代码：

query = 'SELECT text_column FROM `bigquery.table.goes.here` '
lines = p | 'ReadFromBigQuery' >> beam.io.Read(beam.io.BigQuerySource(query=query, use_standard_sql=True))

当我重新运行管道时，出现错误：“WARNING:root:A task failed with exception. expected string or buffer [while running 'Split']”

我认识到“拆分”操作需要一个字符串，但显然没有得到一个字符串。如何修改“ReadFromBigQuery”以使其传递字符串/缓冲区？我是否需要提供表架构或其他东西来将“ReadFromBigQuery”的结果转换为字符串缓冲区？

Answer 1

这是因为BigQuerySource 返回字典的PCollection (dict)，其中字典中的每个键代表一列。对于您的情况，最简单的做法就是在beam.io.Read(beam.io.BigQuerySource(query=query, use_standard_sql=True) 之后应用beam.Map，如下所示：

lines = (p 
|"ReadFromBigQuery" >> beam.io.Read(beam.io.BigQuerySource(query=query, use_standard_sql=True)) 
| "Extract text column" >>  beam.Map(lambda row: row.get("text_column"))
         )

如果您遇到列名问题，请尝试将其更改为u"text_column"。

或者您可以修改您的 Split 转换以提取那里的列值：

'Split' >> (beam.FlatMap(lambda x: re.findall(r'[A-Za-z\']+', x.get("text_column")))
                      .with_output_types(unicode))