google-cloud-dataflow - 在Apache Beam中寻找与文件模式匹配的新文件

Question

我在GCS或其他受支持的文件系统上有一个目录，外部进程正在将新文件写入该目录中。

我想编写一个Apache Beam流传输管道，该管道连续监视此目录中的新文件，并在每个新文件到达时对其进行读取和处理。这可能吗？

Answer 1

从Apache Beam 2.2.0开始，这是可能的。一些API支持此用例:

如果您使用的是TextIO或AvroIO，那么他们将通过TextIO.read().watchForNewFiles()和readAll()上的明确支持此功能，例如:

PCollection<String> lines = p.apply(TextIO.read()
    .from("gs://path/to/files/*")
    .watchForNewFiles(
        // Check for new files every 30 seconds
        Duration.standardSeconds(30),
        // Never stop checking for new files
        Watch.Growth.<String>never()));

如果您使用不同的文件格式，则可以将支持相同API的 FileIO.match().continuously()和 FileIO.matchAll().continuously()与 FileIO.readMatches()结合使用。

API支持指定检查新文件的频率以及何时停止检查(支持的条件例如:“如果在给定时间内没有新输出出现”，“观察到N个输出之后”，“自开始检查以来给定时间之后”)。 ”及其组合)。

请注意，目前此功能目前仅在直接运行器和数据流运行器中有效，并且仅在Java SDK中有效。通常，它将在支持Splittable DoFn的任何运行程序中运行(请参阅 capability matrix)。