Apache Beam File IO - Следите за изменениями файлов

Question

У меня есть компонент потока данных, который следит за новыми файлами в моей папке. Наряду с выпуском новых файлов, я бы хотел, чтобы он испускался, если какой-либо из существующих файлов тоже будет изменен. Есть ли способ добиться этого?

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

Если я помещу новый файл в папку, мой поток данных будет генерировать эти изменения. Но если я изменю существующий файл, это не так. Что я должен сделать, чтобы FileIO.match следил за изменениями файлов?

Answer 1

Что вы используете FileIO.match()

Создает набор сопоставленных ресурсов (как файлов, так и каталогов) в виде MatchResult.Metadata.

...

Возвращенные MatchResult. Метаданные дедуплицируются по имени файла. Например, если это преобразование наблюдает файл с одним и тем же именем несколько раз с разными метаданными (например, из-за того, что файл растет), он выдаст метаданные при первом обнаружении этого файла и проигнорирует будущие изменения в этом файле.

Таким образом, в этом случае вы никогда не сможете увидеть, есть ли изменения в файлах, потому что вы получите только первую запись MatchResult.Metadata для имени файла, дополнительные метаданные игнорируются этот метод.

[EDIT]: FileIO.matchAll()

Ресурсы не дедуплицированы между файловыми шаблонами.

Подобное поведение что FileIO.match() при использовании одного файлового шаблона.

Вы можете использовать FileIO.readMatches()

Преобразует каждый результат match () или matchAll () в FileIO.ReadableFile, который может быть используется для чтения содержимого каждого файла, опционально распаковывая его.

Получив FileIO.ReadableFile, вы можете реализовать logi c, чтобы проверить, был ли файл изменен. Один из вариантов - использовать FileIO.ReadableFile.getMetadata() и сравнить эти MatchResult.Metadata с данными из FileIO.match()

Документов FileIO

Документов FileIO.ReadableFile

В качестве альтернативы вы можете напрямую использовать объект FileIO.Match и выяснить, получаете ли вы необходимое поведение с помощью переопределения или настройки собственной конфигурации.

Документы FileIO. Match