Apache Beam：在启动 ImportTransform Dataflow 模板之前等待 AvroIO 写入步骤完成

Question

我正在使用 Apache Beam 创建一个管道，其中基本上读取 InputFile，转换为 Avro，将 AvroFile 写入存储桶，然后使用 Dataflow 模板将这些 avro 文件导入到 Spanner

我面临的问题是最后一步（将 Avro 文件导入数据库）在前一步（将 Avro 文件写入存储桶）完成之前开始。

我尝试添加 Wait.on 函数，但这仅在返回 PCollection 时有效，但当我将文件写入 avro 时，它会返回 PDone。

代码示例：

// Step 1: Read Files
PCollection<String> lines = pipeline.apply("Reading Input Data exported from Cassandra",TextIO.read().from(options.getInputFile()));

// Step 2: Convert to Avro
lines .apply("Write Item Avro File",AvroIO.writeGenericRecords(spannerItemAvroSchema).to(options.getOutput()).withSuffix(".avro"));

// Step 3: Import to the DataBase

pipeline.apply( new ImportTransform(
                spannerConfig,
                options.getInputDir(),
                options.getWaitForIndexes(),
                options.getWaitForForeignKeys(),
                options.getEarlyIndexCreateFlag()));

同样，问题在于第 3 步在第 2 步完成之前开始

有什么想法吗？

Answer 1

这是 API 中的一个缺陷，请参阅，例如recent discussion 在梁开发列表上。目前唯一的解决方案是 fork AvroIO 以返回 PCollection 或按顺序运行两个管道。