如何在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 = ...(takes a filename, reads the file and produces records)...
    p.apply(Create.of(filenames))
     .apply(ParDo.of(readFile))
     .apply(the rest of your pipeline)

这有一个缺点,即动态工作重新平衡功能不会特别好用,因为它们目前仅适用于Read PTransform的级别,但不适用于具有高扇出的ParDo级别(如此处的那个,它会读取提交并生成所有记录); 并行化仅适用于文件级别,但文件不会拆分为子范围.在阅读莎士比亚的规模上,这不是一个问题,但如果你正在阅读一组截然不同的文件,一些非常大,那么它可能会成为一个问题.

• 实现你自己FileBasedSource(的javadoc,一般的文件),这将返回的类型类似的记录Pair<String, T>,其中String的文件名和T为您正在阅读的记录.在这种情况下,框架将为您处理文件模式匹配,动态工作重新平衡将工作得很好,但是由您来编写读取逻辑FileBasedReader.

这两种解决方法都不理想,但根据您的要求,其中一种可能会为您解决问题.