Apache Beam File IO - 监视文件修改答案

【问题标题】：Apache Beam File IO - Watch for file modificationsApache Beam File IO - 监视文件修改
【发布时间】：2020-04-27 13:05:50
【问题描述】：

我有一个数据流组件，用于监视文件夹中的新文件。除了发出新文件之外，如果任何现有文件也被修改，我希望它发出。有没有办法做到这一点？

PCollection<FileIO.ReadableFile> tfConfigYamls = pipeline.apply(
                "ReadConfigYamls",
                FileIO.match()
                        .filepattern("gs://folder/*.yml")
                        .continuously(Duration.standardMinutes(2), never()))
                .apply(FileIO.readMatches());

如果我在文件夹中放置一个新文件，我的数据流会发出这些更改。但是，如果我修改现有文件，则不会。我应该怎么做才能让 FileIO.match 监视文件修改？

【问题讨论】：

我不确定是否跟踪修改，仅跟踪新文件。 beam.apache.org/releases/javadoc/2.3.0/index.html?org/apache/…
你总是可以实现一个额外的逻辑层来检查文件的任何散列是否发生变化，这可能会触发进一步的分析

标签： google-cloud-platform apache-beam dataflow

【解决方案1】：

你在用什么FileIO.match()

生成匹配资源的集合（文件和目录）作为 MatchResult.Metadata。

...

返回的 MatchResult.Metadata 按文件名进行重复数据删除。为了例如，如果这个变换观察到一个同名的文件有几个具有不同元数据的时间（例如，因为文件正在增长），它将在第一次观察此文件时发出元数据，并将忽略此文件的未来更改。

所以在这种情况下，您将永远无法看到文件是否发生变化，因为您将仅获得文件名的MatchResult.Metadata 的第一条记录，此方法会忽略附加的元数据。

[编辑]： FileIO.matchAll()

文件模式之间不会对资源进行重复数据删除。

与FileIO.match() 使用单个文件模式时的行为非常相似。

你可以使用FileIO.readMatches()

将 match() 或 matchAll() 的每个结果转换为 FileIO.ReadableFile 可用于读取每个文件的内容，可选解压。

获得FileIO.ReadableFile 后，您可以执行逻辑来验证文件是否已被修改。一种选择是使用FileIO.ReadableFile.getMetadata() 并将这个 MatchResult.Metadata 与来自FileIO.match() 的那个进行比较

Docs FileIO

Docs FileIO.ReadableFile

作为替代选项，您可以直接使用 FileIO.Match 对象，并探索是否通过覆盖或设置自己的配置来获得所需的行为。

Docs FileIO.Match

【讨论】：