读取 csv 文件并将数据填充到 BigQuery 表中

Question

以下是应该从 csv 文件读取并写入另一个 csv 文件和 BigQuery 的代码：

import argparse
import logging
import re
import apache_beam as beam
from apache_beam.io import ReadFromText
from apache_beam.io import WriteToText
from apache_beam.metrics import Metrics
from apache_beam.metrics.metric import MetricsFilter
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
parser = argparse.ArgumentParser()
parser.add_argument('--input',
                  dest='input',
                  default='gs://dataflow-samples/shakespeare/kinglear.txt',
                  help='Input file to process.')
parser.add_argument('--output',
                  dest='output',
                  required=True,
                  help='Output file to write results to.')
known_args, pipeline_args = parser.parse_known_args(None)
pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(SetupOptions).save_main_session = True
p = beam.Pipeline(options=pipeline_options)
# Read the text file[pattern] into a PCollection.
lines = p | 'read' >> ReadFromText(known_args.input)
lines | beam.Map(lambda x: x.split(','))
lines | 'write' >> WriteToText(known_args.output)
lines | 'write2' >> beam.io.Write(beam.io.BigQuerySink('xxxx:yyyy.aaaa'))
# Actually run the pipeline (all operations above are deferred).
result = p.run()

它能够写入输出文件，但无法写入 BigQuery 表 (xxxx:yyyy.aaaa)

以下是出现的消息：

WARNING:root:A task failed with exception.
'unicode' object has no attribute 'iteritems'

即使架构相同且 BigQuery 表为空，也不会将 csv 文件中包含的表写入 BigQuery。我怀疑这是因为必须将数据转换为 JSON 格式。 为了使其正常工作，必须对此代码进行哪些更正？您能否提供我必须添加的代码行才能使其正常工作？

Answer 1

查看以下几行：

1: lines = p | 'read' >> ReadFromText(known_args.input)
2: lines | beam.Map(lambda x: x.split(','))
3: lines | 'write' >> WriteToText(known_args.output)
4: lines | 'write2' >> beam.io.Write(beam.io.BigQuerySink('xxxx:yyyy.aaaa'))

1. 将lines 定义为从文本文件中读取的行的PCollection。
2. 通过拆分每一行来创建一个新的单词 PCollection。但它实际上并没有保留那个 PCollection，所以它实际上什么都不做。
3. 将原始行写入文本文件（因此您不会在每行看到一个单词，而是在每个输出中看到一个原始行）。
4. 将从输入读取的行写入 BigQuery 文件。

如果您查看BigQuery tornadoes example，您会发现 (1) 您需要将每一行转换为一个字典，其中包含每个广告列的字段 (2) 您需要提供与该字典匹配的架构到 BigQuerySink。例如：

def to_table_row(x):
  values = x.split(',')
  return { 'field1': values[0], 'field2': values[1] } 

lines = p | 'read' >> ReadFromText(known_args.input)
lines
  | 'write' >> WriteToText(known_args.output)
lines
  | 'ToTableRows' >> beam.Map(to_table_row)
  | 'write2' >> beam.io.Write(beam.io.BigQuerySink(
      'xxxx:yyyy.aaaa',
      schema='field1:INTEGER, field2:INTEGER'))