如何捕捉 BigQueryIO.Write 抛出的异常并挽救输出失败的数据？答案

【问题标题】：How to catch any exceptions thrown by BigQueryIO.Write and rescue the data which is failed to output?如何捕捉 BigQueryIO.Write 抛出的异常并挽救输出失败的数据？
【发布时间】：2018-06-08 16:26:29
【问题描述】：

我想从 Cloud Pub/Sub 读取数据并使用 Cloud Dataflow 将其写入 BigQuery。每个数据都包含一个表 ID，数据本身将保存在其中。

写入 BigQuery 失败的因素有多种：

表格 ID 格式错误。
数据集不存在。
数据集不允许管道访问。
网络故障。

当其中一个故障发生时，流式作业将重试该任务并停止。我尝试使用WriteResult.getFailedInserts() 来挽救不良数据并避免停滞，但效果不佳。有什么好办法吗？

这是我的代码：

public class StarterPipeline {
  private static final Logger LOG = LoggerFactory.getLogger(StarterPipeline.class);

  public class MyData implements Serializable {
    String table_id;
  }

  public interface MyOptions extends PipelineOptions {
    @Description("PubSub topic to read from, specified as projects/<project_id>/topics/<topic_id>")
    @Validation.Required
    ValueProvider<String> getInputTopic();
    void setInputTopic(ValueProvider<String> value);
  }

  public static void main(String[] args) {
    MyOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().as(MyOptions.class);

    Pipeline p = Pipeline.create(options);

    PCollection<MyData> input = p
        .apply("ReadFromPubSub", PubsubIO.readStrings().fromTopic(options.getInputTopic()))
        .apply("ParseJSON", MapElements.into(TypeDescriptor.of(MyData.class))
            .via((String text) -> new Gson().fromJson(text, MyData.class)));
    WriteResult writeResult = input
        .apply("WriteToBigQuery", BigQueryIO.<MyData>write()
            .to(new SerializableFunction<ValueInSingleWindow<MyData>, TableDestination>() {
              @Override
              public TableDestination apply(ValueInSingleWindow<MyData> input) {
                MyData myData = input.getValue();
                return new TableDestination(myData.table_id, null);
              }
            })
            .withSchema(new TableSchema().setFields(new ArrayList<TableFieldSchema>() {{
              add(new TableFieldSchema().setName("table_id").setType("STRING"));
            }}))
            .withFormatFunction(new SerializableFunction<MyData, TableRow>() {
              @Override
              public TableRow apply(MyData myData) {
                return new TableRow().set("table_id", myData.table_id);
              }
            })
            .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
            .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
            .withFailedInsertRetryPolicy(InsertRetryPolicy.neverRetry()));
    writeResult.getFailedInserts()
        .apply("LogFailedData", ParDo.of(new DoFn<TableRow, TableRow>() {
          @ProcessElement
          public void processElement(ProcessContext c) {
            TableRow row = c.element();
            LOG.info(row.get("table_id").toString());
          }
        }));

    p.run();
  }
}

【问题讨论】：

标签： google-bigquery google-cloud-dataflow apache-beam

【解决方案1】：

在管道定义中写入输出时，没有简单的方法来捕获异常。我想您可以通过为 BigQuery 编写自定义 PTransform 来做到这一点。但是，没有办法在 Apache Beam 中本地执行此操作。我也建议不要这样做，因为它会破坏 Cloud Dataflow 的自动重试功能。

在您的代码示例中，您将失败的插入重试策略设置为永不重试。您可以将策略设置为始终重试。这仅在间歇性网络故障（第 4 个要点）等情况下有效。

.withFailedInsertRetryPolicy(InsertRetryPolicy.alwaysRetry())

如果表 ID 格式不正确（第一个要点），那么 CREATE_IF_NEEDED create disposition 配置应该允许 Dataflow 作业自动创建新表而不会出错，即使表 ID不正确。

如果数据集不存在或数据集存在访问权限问题（第 2 和第 3 个要点），那么我认为流式传输作业应该停止并最终失败。没有人工干预，在任何情况下都无法进行。

【讨论】：