Google Cloud 数据流：从具有动态文件名的文件中读取

Question

我正在尝试在 Google Cloud Dataflow 上构建一个可以执行以下操作的管道：

• 收听 Pubsub 订阅中的事件
• 从事件文本中提取文件名
• 读取文件（来自 Google Cloud Storage 存储分区）
• 将记录存储在 BigQuery 中

以下是代码：

Pipeline pipeline = //create pipeline
pipeline.apply("read events", PubsubIO.readStrings().fromSubscription("sub"))
        .apply("Deserialise events", //Code that produces ParDo.SingleOutput<String, KV<String, byte[]>>)
        .apply(TextIO.read().from(""))???

我正在为第三步而苦苦挣扎，不太确定如何访问第二步的输出并在第三步中使用它。我尝试编写产生以下内容的代码：

private ParDo.SingleOutput<KV<String, byte[]>, TextIO.Read> readFile(){
    //A class that extends DoFn<KV<String, byte[]>, TextIO.Read> and has TextIO.read wrapped into processElement method
}

但是，我无法在后续步骤中读取文件内容。

谁能告诉我在第 3 步和第 4 步中我需要写什么，以便我可以逐行使用文件并将输出存储到 BigQuery（或只是记录它）。

Answer 1

表达您的阅读的自然方式是使用TextIO.readAll() 方法，该方法从文件名的输入PCollection 中读取文本文件。此方法已在 Beam 代码库中引入，但目前尚未发布。它将包含在 Beam 2.2.0 版本和相应的 Dataflow 2.2.0 版本中。

Answer 2

您可以使用 SerializableFunction 完成此操作。

你可以的

pipeline.apply(TextIO.read().from(new FileNameFn()));

public class FileNameFn implements SerializableFunction<inputFileNameString, outputQualifiedFileNameStringWithBucket>

显然您可以在通过构造函数参数创建此类实例时静态传递存储桶名称和其他参数。

希望这会有所帮助。