google-cloud-dataflow - 在 google-cloud-dataflow 中使用文件模式匹配时如何获取文件名

Question

有人知道在 google-cloud-dataflow 中使用文件模式匹配时如何获取文件名吗？

我是使用数据流的新手。使用文件模式匹配时如何获取文件名，以这种方式。

p.apply(TextIO.Read.from("gs://dataflow-samples/shakespeare/*.txt"))

我想知道如何检测kinglear.txt、Hamlet.txt等文件名。

Answer 1

如果您想简单地展开文件模式并获得与之匹配的文件名列表，您可以使用 GcsIoChannelFactory.match("gs://dataflow-samples/shakespeare/*.txt") (见 GcsIoChannelFactory)。

如果您想从管道中的 DoFn 下游之一访问“当前文件名” - 目前不支持(尽管有一些解决方法 - 见下文)。这是一个常见的功能请求，我们仍在思考如何以自然、通用和高性能的方式将它最好地融入到框架中。

一些解决方法包括:

编写这样的管道(tf-idf 示例使用这种方法):

DoFn readFile = ...(获取文件名，读取文件并生成记录)...
p.apply(Create.of(filenames))
.apply(ParDo.of(readFile))
.apply(管道的其余部分)

这有一个缺点，即动态工作重新平衡功能不会特别好用，因为它们目前仅适用于读取 PTransform 的级别，而不适用于具有高扇出的 ParDo 级别(就像这里的那个，它会读取一个归档并制作所有记录)；并且并行化仅适用于文件级别，但文件不会拆分为子范围。在阅读莎士比亚的规模上，这不是问题，但如果您正在阅读一组大小迥异的文件，有些文件非常大，那么这可能会成为一个问题。

实现自己的 FileBasedSource ( javadoc , general documentation ) 将返回类似 Pair<String, T> 类型的记录哪里String是文件名和 T是您正在阅读的记录。在这种情况下，框架将为您处理文件模式匹配，动态工作重新平衡会很好地工作，但是您可以在 FileBasedReader 中编写读取逻辑。 .

这两种变通方法都不理想，但根据您的要求，其中一种可能适合您。