是否遇到从数据流管道到BigQuery的低速流式写入？

Question

在使用流式插入和Python SDK2.23写入BigQuery时，我遇到了意外的性能问题。

在没有写入步骤的情况下，流水线在一个工作线程上运行，占用大约20-30%的CPU。添加BigQuery步骤，流水线可以扩展到6个工作进程，所有工作进程都占用70-90%的CPU。

我对数据流和波束很陌生，可能这种行为很正常，或者我做错了什么，但在我看来，使用6台机器每秒向BigQuery写入250行数据有点重。我想知道如何才能达到每秒100K行的插入配额。

我的管道看起来像这样：

p
    | "Read from PubSub" >> beam.io.ReadFromPubSub(subscription=options.pubsub_subscription) # ~40/s
    | "Split messages" >> beam.FlatMap(split_messages) # ~ 400/s
    | "Prepare message for BigQuery" >> beam.Map(prepare_row)
    | "Filter known message types" >> beam.Filter(filter_message_types) # ~ 250/s
    | "Write to BigQuery" >> beam.io.WriteToBigQuery(
          table=options.table_spec_position,
          schema=table_schema,
          write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
          create_disposition=beam.io.BigQueryDisposition.CREATE_NEVER,
          additional_bq_parameters=additional_bq_parameters,
      )

流水线使用这些选项运行，尽管我在没有使用流引擎的情况下遇到了类似的行为。

--enable_streaming_engine \
--autoscaling_algorithm=THROUGHPUT_BASED \
--max_num_workers=15 \
--machine_type=n1-standard-2 \
--disk_size_gb=30 \

指标截图：

​

我的问题是，这种行为是否正常，或者我可以做些什么来减少这条管道所需的工人数量。谢谢!

更新:这是数据流图的最后一步的图像，显示了墙的时间。(在作业运行1小时后拍摄)。之前的所有其他步骤的挂起时间都非常短，只有几秒钟。

​

​

Answer 1

经过调试，我发现有一些无效的消息无法写入BigQuery (并且没有记录错误)。因此，对于遇到类似问题的任何人：

在将beam.io.WriteToBigQuery的insert_retry_strategy更改为RETRY_NEVER并打印出死信pCollection之后，我修复了格式错误的消息，性能得到了提高。我猜由于RETRY_ALWAYS的默认策略，有一些无效消息被卡住了。