Как определить искомый путь к файлу потокового файла или имя файла?

Question

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

У меня проблема с тем, что выходной файл похож на part-00012-8d701427-8289-41d7-9b4d-04c5d882664d-c000.txt. Это делает невозможным вывод файлов в течение последнего часа.

Можно ли настроить выходной файл в timestamp_xxx или что-то в этом роде? Или я могу выводить по разным путям для каждой партии?

Answer 1

Вы не можете изменить имя сохраненных файлов. Однако вы можете изменить структуру папок, в которой они сохранены. Используйте partitionBy() для разделения данных после указанных столбцов в наборе данных, в этом случае год, месяц, день и час могут представлять интерес:

df.writeStream 
  .format("parquet") // can be "orc", "json", "csv", etc.
  .option("path", "/path/to/save/") 
  .partitionBy("year", "month", "day", "hour") 
  .start() 

Это создаст структуру папок, начиная с path, которая может выглядеть следующим образом:

year=2018
|
|--> month=06
|    |
|    |--> day=26
|    |    |
|    |    |--> hour=10
|    |    |--> hour=11
|    |    |--> ...
|    |
|    |--> day=27
|    |    |
|    |    |--> ...

Конечно, для разделения файлов могут использоваться другие столбцы, в зависимости от того, что доступно.

Answer 2

Я полагаю, что этот формат файла является внутренней вещью, которая используется Spark для сохранения значений для каждого раздела. Если вы используете какое-то хранилище BLOB-объектов (извините, я пользователь Windows), вы все равно должны иметь возможность загрузить файлы обратно из места вывода, а затем снова поработать с ними, используя DataFrame.

То, что я пытаюсь сказать, это то, что, хотя у вас мало что сказано в именах файлов, так как это то, что Spark делает сам, это не должно мешать вам создавать свой собственный рабочий процесс, в котором вы пакетируете вещи, в которые вы заглядываете внутрь. файлы для некоторой метки времени (я предполагаю, что содержимое файла имеет какой-то столбец DataTime, если не стоит добавлять его)

Именно так я бы поступил с вещами, сделав временную метку частью содержимого файла, и затем вы можете использовать фактическое содержимое файла (как я говорю, прочитайте в DataFrame, скажем), а затем просто использовать обычные операции DataFrame / Map на загруженные выходные данные

Я вроде бы примерно говорю об этом здесь .