python - 修改 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 返回字典 (dict) 的 PCollection，其中字典中的每个键代表一个列。对于您的情况，最简单的做法是在 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))