Apache Beam 中的动态目的地

Question

我有一个 PCollection [String] 说“X”，我需要转储到 BigQuery 表中。 表目标和它的模式在 PCollection[TableRow] 中说“Y”。 如何以最简单的方式实现这一点？

我尝试从“Y”中提取表和模式并将其保存在静态全局变量（分别为表名和模式）中。但奇怪的是，BigQueryIO.writeTableRows() 总是将变量 tableName 的值设为 null。但它得到了模式。我尝试记录这些变量的值，我可以看到两者的值都存在。

这是我的管道代码：

static String tableName;
static TableSchema schema;

PCollection<String> read = p.apply("Read from input file",
  TextIO.read().from(options.getInputFile()));

PCollection<TableRow> tableRows = p.apply(
  BigQueryIO.read().fromQuery(NestedValueProvider.of(
    options.getfilename(),
    new SerializableFunction<String, String>() {
         @Override
         public String apply(String filename) {
           return "SELECT table,schema FROM `BigqueryTest.configuration` WHERE file='" + filename +"'";
         }
    })).usingStandardSql().withoutValidation());

final PCollectionView<List<String>> dataView = read.apply(View.asList());

tableRows.apply("Convert data read from file to TableRow",
  ParDo.of(new DoFn<TableRow,TableRow>(){
    @ProcessElement
    public void processElement(ProcessContext c) {
      tableName = c.element().get("table").toString();
      String[] schemas = c.element().get("schema").toString().split(",");
      List<TableFieldSchema> fields = new ArrayList<>();
      for(int i=0;i<schemas.length;i++) {
        fields.add(new TableFieldSchema()
          .setName(schemas[i].split(":")[0]).setType(schemas[i].split(":")[1]));
      }
      schema = new TableSchema().setFields(fields);

      //My code to convert data to TableRow format.
    }}).withSideInputs(dataView)); 


tableRows.apply("write to BigQuery", 
  BigQueryIO.writeTableRows()
    .withSchema(schema)
    .to("ProjectID:DatasetID."+tableName)
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

一切正常。只有 BigQueryIO.write 操作失败，我收到错误 TableId is null。

我也尝试使用 SerializableFunction 并从那里返回值，但我仍然得到 null。

这是我尝试过的代码：

tableRows.apply("write to BigQuery",
BigQueryIO.writeTableRows()
  .withSchema(schema)
  .to(new GetTable(tableName))
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

public static class GetTable implements SerializableFunction<String,String> {
  String table;

  public GetTable() {
    this.table = tableName;
  }

  @Override
  public String apply(String arg0) {
    return "ProjectId:DatasetId."+table;
  }
}

我也尝试过使用 DynamicDestinations，但我收到一条错误消息，提示未提供架构。老实说，我是 DynamicDestinations 概念的新手，我不确定自己是否正确。

这是我尝试过的代码：

tableRows2.apply(BigQueryIO.writeTableRows()
  .to(new DynamicDestinations<TableRow, TableRow>() {
    private static final long serialVersionUID = 1L;
    @Override
    public TableDestination getTable(TableRow dest) {
      List<TableRow> list = sideInput(bqDataView); //bqDataView contains table and schema
      String table = list.get(0).get("table").toString();
      String tableSpec = "ProjectId:DatasetId."+table;
      String tableDescription = "";
      return new TableDestination(tableSpec, tableDescription);
    }

    public String getSideInputs(PCollectionView<List<TableRow>> bqDataView) {
      return null;
    }

    @Override
    public TableSchema getSchema(TableRow destination) {
      return schema;   //schema is getting added from the global variable
    }
    @Override
    public TableRow getDestination(ValueInSingleWindow<TableRow> element) {
      return null;
    }
}.getSideInputs(bqDataView)));

请让我知道我做错了什么以及我应该走哪条路。

谢谢。

Answer 1

您遇到问题的部分原因是管道执行的两个阶段。首先，管道是在您的机器上构建的。这是 PTransforms 的所有应用发生的时候。在您的第一个示例中，这是执行以下行的时间：

BigQueryIO.writeTableRows()
  .withSchema(schema)
  .to("ProjectID:DatasetID."+tableName)

但是，ParDo 中的代码会在您的管道执行时运行，并且它在许多机器上都会这样做。所以下面的代码比管道构造运行晚得多：

@ProcessElement
public void processElement(ProcessContext c) {
  tableName = c.element().get("table").toString();
  ...
  schema = new TableSchema().setFields(fields);
  ...
}

这意味着在创建 BigQueryIO 接收器时不会设置 tableName 和架构字段。

您使用 DynamicDestinations 的想法是正确的，但您需要移动代码以将目标模式实际生成到该类中，而不是依赖并非在所有机器上都可用的全局变量。