谷歌数据流：从谷歌存储中的文件读取的 AvroIO 作为运行时参数传递

Question

我想使用 java SDK 2 在我的数据流中读取 Avro 文件
我已经使用基于上传到存储桶的文件触发的云功能来安排我的数据流。

以下是选项代码：

ValueProvider <String> getInputFile();
void setInputFile(ValueProvider<String> value);

我正在尝试使用以下代码读取此输入文件：

PCollection<user> records = p.apply(
    AvroIO.read(user.class)
    .from(String.valueOf(options.getInputFile())));

运行管道时出现以下错误：

java.lang.IllegalArgumentException: Unable to find any files matching RuntimeValueProvider{propertyName=inputFile, default=gs://test_bucket/user.avro, value=null}

同样的代码在 TextIO 的情况下也能正常工作。
我们如何读取上传的 Avro 文件以触发触发数据流管道的云功能？

Answer 1

请尝试...from(options.getInputFile()))，不要将其转换为字符串。

为简单起见，您甚至可以将选项定义为简单字符串：

   String getInputFile();
   void setInputFile(String value);

Answer 2

您只需使用from(options.getInputFile())：AvroIO 明确支持从ValueProvider 读取。

当前代码采用options.getInputFile()，这是一个ValueProvider，在其上调用JavatoString()函数，它提供了一个人类可读的调试字符串"RuntimeValueProvider{propertyName=inputFile, default=gs://test_bucket/user.avro, value=null}"并将其作为文件名传递给AvroIO以读取，当然这个字符串不是一个有效的文件名，这就是代码当前不起作用的原因。

还请注意，ValueProvider 的全部意义在于它是一个占位符，用于在构造管道时未知并将稍后提供（可能管道将执行多次，提供不同的值） - 因此在管道施工时提取ValueProvider 的值在设计上是不可能的，因为没有价值。但是在运行时（例如在 DoFn 中），您可以通过调用 .get() 来提取值。