Как я могу идентифицировать обработанные файлы в задании потока данных

Question

Как определить обработанные файлы в задании потока данных?Я использую подстановочный знак для чтения файлов из облачного хранилища.но каждый раз, когда задание запускается, оно перечитывает все файлы.

Это пакетное задание, и ниже приводится пример чтения TextIO, который я использую.

PCollection<String> filePColection  = pipeline.apply("Read files from Cloud Storage ", TextIO.read().from("gs://bucketName/TrafficData*.txt"));

Answer 1

Чтобы просмотреть список файлов, соответствующих вашему шаблону, вы можете использовать gsutils, который является утилитой командной строки Cloud Storage.Вы бы сделали следующее:

gsutils ls gs://bucketName/TrafficData*.txt

---

Теперь, когда нужно выполнить пакетное задание несколько раз, ваш конвейер не сможет узнать, какие файлы он уже проанализировал или нет.Чтобы избежать анализа новых файлов, вы можете выполнить одно из следующих действий:

1. Определить задание потоковой передачи и использовать функцию TextIO watchForNewFiles.Вы должны были бы оставить свою работу для выполнения до тех пор, пока вы хотите продолжать обработку файлов.

2. Найдите способ предоставить вашему конвейеру файлы, которые уже были проанализированы.Для этого каждый раз, когда вы запускаете свой конвейер, вы можете сгенерировать список файлов для анализа, поместить его в PCollection, прочитать каждый с TextIO.readAll() и где-то сохранить список проанализированных файлов.Позже, когда вы снова запустите конвейер, вы можете использовать этот список в качестве черного списка для файлов, которые вам не нужно запускать снова.

Дайте мне знать в комментариях, если вы хотитевыработать решение вокруг одного из этих двух вариантов.