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

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

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

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


apache-spark hdfs spark-structured-streaming
person skyer CC    schedule 27.06.2018    source источник

Ответы (2)


arrow_upward
0
arrow_downward

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

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

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

Я как бы грубо говорю об этом здесь.

person sacha barber    schedule 27.06.2018
comment
Моя проблема не в том, чтобы обрабатывать все данные по пути вывода, потому что вывод файлов продолжается (около 100 файлов на пакет). Я хочу объединять эти выходные данные в один каталог ежечасно, используя map / reduce. Это время пакетной обработки и количество повторных разделов нельзя изменить, поскольку ресурсы ограничены. Это позволяет выводить более 1000 файлов в час и делает входной аргумент hadoop слишком длинным, поскольку мне приходится объединять каждый файл один за другим в строку. - person skyer CC; 27.06.2018
comment
Я думаю, что то, что я показываю здесь, и то, что показано ниже @shaido, указывает на то, как вы можете создавать файлы с каким-то собственным именем, а затем вы сможете использовать синтаксис glob, как описано здесь: stackoverflow.com/questions/31782763/. Это позволит вам загружать ТОЛЬКО ту партию, которую вы хотите. Это руководство показывает вам: - person sacha barber; 27.06.2018
comment
Использование синтаксиса glob позволит вам сделать что-то вроде этого sc.textFile("/user/Orders/2015072[7-9]*,/user/Orders/2015073[0-1]*") - person sacha barber; 27.06.2018
comment
Спасибо за совет, попробую воспользоваться методом @haido, посмотрим, что получится. - person skyer CC; 27.06.2018

arrow_upward
7
arrow_downward

Вы не можете изменить название сохраненных файлов. Однако вы можете изменить структуру папок, в которой он хранится. Используйте 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
|    |    |
|    |    |--> ...

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

person Shaido    schedule 27.06.2018
comment
Возможно ли использование формата текстового файла для этого, например `.format (text) .option (path, some hdfs path) .partitionBy ()` - person skyer CC; 27.06.2018
comment
@skyerCC: Да, можно. Попробуйте изменить parquet на csv, и вы получите столбцы, разделенные запятыми, в текстовом файле. - person Shaido; 27.06.2018