我正试图用Dataflow创建一个流式管道，该管道从PubSub主题读取消息，最终将消息写入BigQuery表。我不想使用任何数据流模板。
目前，我只想在从Google VM实例执行的Python3脚本中创建一个管道，对从Pubsub到达的每个消息执行加载和转换过程(解析其中包含的记录并添加一个新字段(，最终将结果写入BigQuery表。
简单来说，我的代码是：
#!/usr/bin/env python
from apache_beam.options.pipeline_options import PipelineOptions
from google.cloud import pubsub_v1,
import apache_beam as beam
import apache_beam.io.gcp.bigquery
import logging
import argparse
import sys
import json
from datetime import datetime, timedelta
def load_pubsub(message):
try:
data = json.loads(message)
records = data["messages"]
return records
except:
raise ImportError("Something went wrong reading data from the Pub/Sub topic")
class ParseTransformPubSub(beam.DoFn):
def __init__(self):
self.water_mark = (datetime.now() + timedelta(hours = 1)).strftime("%Y-%m-%d %H:%M:%S.%f")
def process(self, records):
for record in records:
record["E"] = self.water_mark
yield record
def main():
table_schema = apache_beam.io.gcp.bigquery.parse_table_schema_from_json(open("TableSchema.json"))
parser = argparse.ArgumentParser()
parser.add_argument('--input_topic')
parser.add_argument('--output_table')
known_args, pipeline_args = parser.parse_known_args(sys.argv)
with beam.Pipeline(argv = pipeline_args) as p:
pipe = ( p | 'ReadDataFromPubSub' >> beam.io.ReadStringsFromPubSub(known_args.input_topic)
| 'LoadJSON' >> beam.Map(load_pubsub)
| 'ParseTransform' >> beam.ParDo(ParseTransformPubSub())
| 'WriteToAvailabilityTable' >> beam.io.WriteToBigQuery(
table = known_args.output_table,
schema = table_schema,
create_disposition = beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
write_disposition = beam.io.BigQueryDisposition.WRITE_APPEND)
)
result = p.run()
result.wait_until_finish()
if __name__ == '__main__':
logger = logging.getLogger().setLevel(logging.INFO)
main()
(例如(PubSub主题中发布的消息通常如下所示：
'{"messages":[{"A":"Alpha", "B":"V1", "C":3, "D":12},{"A":"Alpha", "B":"V1", "C":5, "D":14},{"A":"Alpha", "B":"V1", "C":3, "D":22}]}'
如果在记录中添加了字段"E"，那么记录的结构(Python中的字典(和字段的数据类型就是BigQuery表所期望的。
我想处理的问题是：
如果一些消息带有意外的结构，我想分叉管道扁平化，并将它们写入另一个BigQuery表中。
如果某些消息带有意外的字段数据类型，那么在管道的最后一级，当它们应该写入表时，将发生错误。我想通过将记录转移到第三个表来管理这种类型的错误。
我阅读了以下页面上的文档，但一无所获：https://cloud.google.com/dataflow/docs/guides/troubleshooting-your-pipelinehttps://cloud.google.com/dataflow/docs/guides/common-errors
顺便说一句，如果我选择通过从PubSubSubscription读取并写入BigQuery的模板来配置管道的选项，我会得到以下模式，它与我正在寻找的模式相同：
模板：BigQuery 的云发布/子订阅
您无法捕获BigQuery接收器中发生的错误。您在bigquery中写入的消息必须是好的。
最好的模式是执行一个检查消息结构和字段类型的转换。在出现错误的情况下，您创建一个错误流，并将此问题流写入一个文件(例如，或在没有模式的表中，以纯文本形式写入消息(
当BigQuery接收器出现错误时，我们会执行以下操作。
- 向GCP错误报告发送一条消息(无堆栈争用(，以便通知开发人员
- 将错误记录到StackDriver
- 停止管道执行(在开发人员解决问题之前，消息等待的最佳位置是插入pubSub订阅(