Apache Beam TextIO.ReadAll 如何发出 KeyValue 而不是 Pcollection 的字符串

Question

管道从 PUBSUBIo 读取开始。 PubSub IO 中的消息是 GCS 文件路径。我知道我可以使用ReadAll() 从每个路径发出行。但是，它不符合我的目的（有关文件路径的信息丢失）。我需要的是发出一个KV<'Filepath','Lines inside files'>。

PubSUB 消息看起来像

Message1 -> gs://folder1/Topic1/topicfile1.gz
Message2 -> gs://folder1/Topic2/topicfile2.gz

假设文件内容如下

topicfile1.gz
{
topic1.line1
topic1.line2
}

topicfile2.gz
{
topic2.line1
topic2.line2
}

我期待的是一个像下面这样的 pcollection

{KV<'gs://folder1/Topic1/topicfile1.gz','topic1.line1'>}
{KV<'gs://folder1/Topic1/topicfile1.gz','topic1.line2'>}
{KV<'gs://folder1/Topic2/topicfile2.gz','topic2.line1'>}
{KV<'gs://folder1/Topic2/topicfile2.gz','topic2.line2'>}

我找不到从 ParDo 函数内的路径读取文件以将路径映射到行的方法。

希望这很清楚。

Answer 1

如果我正确理解问题，我认为 TextIO 开箱即用不支持此功能。

详情

当您应用 readAll() 之类的转换时，在从 IO 获取初始文件路径和最终从所有文件发出所有行之间涉及几个步骤。

例如逻辑in TextIO:

• 它接受PCollection 的文件路径（或路径模式）；
• 它应用FileIO.matchAll()，将路径模式的PCollection 转换为描述这些路径的MatchResult.Metadata 对象的PCollection；
• 然后它应用FileIO.readMatches() 将元数据对象转换为描述特定文件的ReadableFile 对象；
• 最后它应用TextIO.readFiles() 接受ReadableFile 并输出该文件中的所有字符串；
  • 在这最后一步中，您可能希望在输出中添加一个文件路径，以便您知道哪个字符串来自哪个文件。如果可以选择将最后一步更改为发出KV<ReadableFile, String> 而不仅仅是字符串，这样您就可以使用ReadableFile.metadata 访问文件路径，这会有什么帮助。

环顾该代码，似乎从文件中发出原始行是目前使用TextIO 唯一受支持的处理方式。

解决方法

可能最直接的方法是编写自己的PTransform，类似于TextIO.ReadAll。这将像这样工作：

高级：

• 创建和自定义您自己的版本TextIO.ReadAll；
• ReadAllViaFileBasedSource;
• 更改您的 ReadAllViaFileBasedSource 版本以发出您想要的内容；
• 使用这个自定义版本的TextIO.ReadAll，它使用你的自定义版本ReadAllViaFileBasedSource，它会发出正确的东西；

稍微详细一点：

• 只需复制整个TextIO.ReadAll，它是我上面提到的FileIO that implements the steps 的一个非常短的包装；
• 但是在expand()、at the last step，而不是readFiles() 中，您将应用自定义逻辑来发出您想要的KVs：
  • readFiles() 现在是 implemented by ReadAllViaFileBasedSource;
  • ReadAllViaFileBasedSource 似乎是 converts ReadableFiles into strings 的真实存在；
  • 您将创建ReadAllViaFileBasedSource 的副本并更改输出逻辑，这样不仅会发出文件，还会发出元数据；