Как указать несколько путей ввода к заданию потока данных

Question

Я хочу запустить задание потока данных на нескольких входах из облачного хранилища Google, но пути, которые я хочу передать в задание, не могут указываться только с помощью оператора glob *.

Рассмотрим эти пути:

gs://bucket/some/path/20160208/input1 
gs://bucket/some/path/20160208/input2 
gs://bucket/some/path/20160209/input1 
gs://bucket/some/path/20160209/input2 
gs://bucket/some/path/20160210/input1 
gs://bucket/some/path/20160210/input2 
gs://bucket/some/path/20160211/input1 
gs://bucket/some/path/20160211/input2 
gs://bucket/some/path/20160212/input1 
gs://bucket/some/path/20160212/input2 

Я хочу, чтобы моя работа, чтобы работать над файлами в 20160209, 20160210 и 20160211 каталогов, но не на 20160208 (первый) и 20160212 (последний). На самом деле существует намного больше дат, и я хочу иметь возможность указать произвольный диапазон дат для моей работы.

The docs for TextIO.Read говорят:

Стандартный Java Filesystem GLOB модели ("*", "[..]" "?") Поддерживаются.

Но я не могу заставить это работать. Есть ссылка на Java Filesystem glob patterns, которая, в свою очередь, ссылается на getPathMatcher(String), в которой перечислены все варианты подстановки. Один из них - {a,b,c}, который выглядит точно так же, как мне нужно, если я передаю gs://bucket/some/path/201602{09,10,11}/* в TextIO.Read#from. Я получаю «Невозможно развернуть шаблон файла».

Может быть, документы означают, что только*, ? и […] поддерживаются, и если это так, то как я могу построить Glob, что Dataflow будет принимать и что может соответствовать произвольный диапазон дат, как один я описываю выше?

Update: Я понял, что я могу написать кусок кода, так что я могу передать в пути префиксы, через запятую, создать входной сигнал от каждого и использовать Flatten преобразования, но это похоже на очень неэффективный способ сделать это. Похоже, что первый шаг читает все входные файлы и сразу же записывает их снова во временное место на GCS. Только когда все входы были прочитаны и записаны, начинается фактическая обработка. Этот шаг совершенно не нужен в работе, которую я пишу. Я хочу, чтобы задание читало первый файл, начинал его обрабатывать и читал дальше, и так далее. Это просто вызвало тонну других проблем, я постараюсь заставить его работать, но он чувствует себя как тупик из-за первоначальной перезаписи.

Answer 1

Документы действительно означают, что поддерживаются только *, ? и [...]. Это означает, что произвольные подмножества или диапазоны в алфавитном или цифровом порядке не могут быть выражены как один глобус.

Вот некоторые подходы, которые могли бы работать для вас:

1. Если дата, представленная в пути файла также присутствует в записях в файлах, то самое простое решение, чтобы прочитать их все и использовать Filter transform, чтобы выбрать интересующий вас диапазон дат.
2. Подход, который вы пробовали из многих чтений в разделяющих TextIO.Read, преобразует и выравнивает их в порядке для небольших наборов файлов; наш tf-idf example делает это.Вы можете выразить произвольные числовые диапазоны с небольшим числом комков поэтому это не должны быть один для чтения в файл (например, диапазон два символа «23 через 67» является 2[3-] плюс [3-5][0-9] плюс 6[0-7])
3. Если подмножество файлов более произвольно, тогда число глобусов/имен файлов может превышать максимальный размер графика, а последняя рекомендация - перечислить список файлов в PCollection и использовать преобразование ParDo для чтения каждого файла и извлечения его содержимого.

Надеюсь, это поможет!