您可以通过
WriteToText添加
.csv后缀和来实现
headers。考虑到您需要将查询结果解析为CSV格式。作为示例，我使用了莎士比亚公共数据集和以下查询：
从
bigquery-public-data.samples.shakespeare中选择单词，单词数，语料库，其中CHAR_LENGTH（word）> 3按单词数计数排序限制10
现在，我们通过以下方式读取查询结果：
BQ_DATA = p | 'read_bq_view' >> beam.io.Read( beam.io.BigQuerySource(query=query, use_standard_sql=True))
BQ_DATA现在包含键值对：
{u'corpus': u'hamlet', u'word': u'HAMLET', u'word_count': 407}{u'corpus': u'kingrichardiii', u'word': u'that', u'word_count': 319}{u'corpus': u'othello', u'word': u'OTHELLO', u'word_count': 313}我们可以应用一个
beam.Map函数只产生值：
BQ_VALUES = BQ_DATA | 'read values' >> beam.Map(lambda x: x.values())
摘录
BQ_VALUES：
[u'hamlet', u'HAMLET', 407][u'kingrichardiii', u'that', 319][u'othello', u'OTHELLO', 313]
最后再次映射，使所有列值用逗号而不是列表分开（考虑到如果双引号可以出现在字段中，则需要转义双引号）：
BQ_CSV = BQ_VALUES | 'CSV format' >> beam.Map( lambda row: ', '.join(['"'+ str(column) +'"' for column in row]))
现在，我们将结果后缀和标头写入GCS：
BQ_CSV | 'Write_to_GCS' >> beam.io.WriteToText( 'gs://{0}/results/output'.format(BUCKET), file_name_suffix='.csv', header='word, word count, corpus')书面结果：
$ gsutil cat gs://$BUCKET/results/output-00000-of-00001.csvword, word count, corpus"hamlet", "HAMLET", "407""kingrichardiii", "that", "319""othello", "OTHELLO", "313""merrywivesofwindsor", "MISTRESS", "310""othello", "IAGO", "299""antonyandcleopatra", "ANTONY", "284""asyoulikeit", "that", "281""antonyandcleopatra", "CLEOPATRA", "274""measureforemeasure", "your", "274""romeoandjuliet", "that", "270"
