Как обновить файлы HDFS после каждого запуска?

Question

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

Допустим, я начал потоковую передачу, и я добавил файл-1.csv в свой каталог ... ничего не записано в папку HDFS. Затем я добавил файл-2.csv. Тем не менее, он не записывается в HDFS. Затем я добавил файл-3.csv .. на этот раз обработанный файл file-1.csv записывается в HDFS ...

File 1 - no prcess
File 2 - no process
File 3 - process and written file 1 data to HDFS
file4 - process and written file 2 data to HDFS 

Я не уверен, почему это происходит, даже если это структурированная потоковая передача

может кто-нибудь как решить эту проблему?

моя команда вывода приведена ниже:

FetRepo
  .writeStream
  .outputMode("append")
  .partitionBy("data_dt")
  .format("csv")
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .option("path", "hdfs://ffff/apps/hive/warehouse/area.db/fin_repo/")
  .start

Answer 1

Задайте себе вопрос: "Как часто я добавляю файлы?" и как это связано с Trigger.ProcessingTime("10 seconds")? С конфигурацией вы не должны ожидать, что что-то случится в течение 10 секунд.

Еще одна вещь, на которую стоит обратить внимание, это то, что вы используете outputMode("append"), который будет выводить только строки для агрегатов (групп), которые были добавлены с момента последнего триггера.

С Основные понятия :

Режим добавления - во внешнюю память будут записаны только новые строки, добавленные в таблицу результатов с момента последнего запуска.

Обратите внимание, что (цитируя тот же документ): "Это применимо только к запросам, в которых не ожидается изменение существующих строк в таблице результатов."