Сохранение набора данных искры в существующий файл csv

Question

Я пытаюсь сохранить содержимое набора данных в csv, используя

df.coalesce(1)
  .write()
  .format("csv")
  .mode("append")
  .save(PATH+"/trial.csv");

Моя цель - продолжать добавлять результаты набора данных в файл trial.csv. Однако он создает папку с именем trial.csv и создает внутри нее csv. Когда я запускаю его снова, он создает еще один файл csv внутри папки trail.csv. Но я просто хочу, чтобы он продолжал добавлять в один файл csv, чего я не могу сделать.

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

Answer 1

Добавление к существующему файлу того, что сложно сделать для распределенного многопоточного приложения, оно превратит распараллеливание в последовательную задачу. Обычно это достигается путем сохранения для каждого потока или задачи в Spark одного файла по указанному пути, и этот путь будет папкой, содержащей все файлы. Чтобы прочитать их, вы можете прочитать всю папку.

Если ваши данные не большие, и вам действительно нужен один файл, попробуйте с помощью метода repartition установить 1, это сделает единственную задачу для записи новые данные, но он никогда не добавит данные в предыдущие файлы.

Вы должны быть осторожны, но вы можете сделать что-то вроде этого:

df.union(spark.read(PATH+"/trial.csv"))
  .coalesce(1)
  .write
  .format("csv")
  .mode("append")
  .save(PATH+"/trial_auxiliar.csv")

Затем переместите его в предыдущую папку , с искрой или командой перемещения Had oop. Никогда не выполняйте запись и чтение в одном и том же задании в одной и той же папке и помните, что это не гарантирует порядок данных.