将特定的 PCollection 写入 BigQuery

Question

假设我创建了两个输出 PCollection 作为 SideOutputs 的结果，并且根据某些条件我只想将其中一个写入 BigQuery。这该怎么做？

基本上我的用例是我试图使 Write_Append 和 Write_Truncate 动态化。我从我在 BigQuery 中维护的配置表中获取信息（追加/截断）。因此，根据我在配置表中的内容，我必须应用 Truncate 或 Append。

因此，使用 SideOutputs 我能够创建两个 PCollections（分别为 Append 和 Truncate），其中一个将为空。并且必须将包含所有行的行写入 BigQuery。这种方法正确吗？

我正在使用的代码：

 final TupleTag<TableRow> truncate =
                  new TupleTag<TableRow>(){};
              // Output that contains word lengths.
              final TupleTag<TableRow> append =
                  new TupleTag<TableRow>(){};

              PCollectionTuple results = read.apply("convert to table row",ParDo.of(new DoFn<String,TableRow>(){
              @ProcessElement
              public void processElement(ProcessContext c)
              {
                  String value = c.sideInput(configView).get(0).toString();
                  LOG.info("config: "+value);
                  if(value.equals("truncate")){
                      LOG.info("outputting to truncate");
                      c.output(new TableRow().set("color", c.element()));
                  }
                  else
                  {
                      LOG.info("outputting to append");
                      c.output(append,new TableRow().set("color", c.element()));
                  }
                  //c.output(new TableRow().set("color", c.element()));
              }
          }).withSideInputs(configView).withOutputTags(truncate,
                  TupleTagList.of(append)));

              results.get(truncate).apply("truncate",BigQueryIO.writeTableRows()
                        .to("projectid:datasetid.tableid")
                        .withSchema(schema)
                        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
                        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

              results.get(append).apply("append",BigQueryIO.writeTableRows()
                        .to("projectid:datasetid.tableid")
                        .withSchema(schema)
                        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

我需要执行两者中的一个。如果我做这两个表无论如何都会被截断。

附：我正在使用 Java SDK (Apache Beam 2.1)

Answer 1

我相信您是对的，如果您的管道完全使用 WRITE_TRUNCATE 写入 BigQuery 表，则当前即使没有数据，该表也会被截断。在这种情况下，请随时 file a JIRA 支持更多可配置的行为。

因此，如果您希望它有条件地不被截断，则需要有条件地根本不包含该写入转换。有没有办法将条件推到那个级别，还是实际上必须从管道中的其他数据计算条件？

（我能想到的唯一解决方法是使用 DynamicDestinations 动态选择要截断的表的名称，并截断其他一些虚拟空表 - 我可以在您对上一段的回答后详细说明）