Как написать некоторый val как файл JSON на S3 с опцией перезаписи? [Дубликат]

df.write.mode ('overwrite'). Паркет («/ output / folder / path») работает, если вы хотите перезаписать файл паркета с помощью python. Это в искру 1.6.2. API может отличаться в более поздних версиях

Документация для параметра spark.files.overwrite говорит следующее: «Записывать ли файлы, добавленные через SparkContext.addFile(), когда целевой файл существует, а его содержимое не соответствует файлам источника». Таким образом, это не влияет на метод saveAsTextFiles.

Вы можете сделать это перед сохранением файла:

val hadoopConf = new org.apache.hadoop.conf.Configuration()
val hdfs = org.apache.hadoop.fs.FileSystem.get(new java.net.URI("hdfs://localhost:9000"), hadoopConf)
try { hdfs.delete(new org.apache.hadoop.fs.Path(filepath), true) } catch { case _ : Throwable => { } }

Aas объяснено здесь: http: // apache-spark- user-list.1001560.n3.nabble.com/How-can-I-make-Spark-1-0-saveAsTextFile-to-overwrite-existing-file-td6696.html

, поскольку df.save(path, source, mode) устарел, ( http://spark.apache.org/docs/1.5.0/api/scala/index.html#org.apache.spark.sql.DataFrame )

использовать df.write.format(source).mode("overwrite").save(path), где df.write является DataFrameWriter

'source' может быть («com.databricks.spark.avro» | «паркет» | «json»)

В документации pyspark.sql.DataFrame.save (в настоящее время в 1.3.1) вы можете указать mode='overwrite' при сохранении DataFrame:

myDataFrame.save(path='myPath', source='parquet', mode='overwrite')

I Мы проверили, что это даже удалит файлы с файлами разделов. Итак, если вы сказали изначально 10 разделов / файлов, но затем перезаписали папку с DataFrame, в которой было только 6 разделов, результирующая папка будет иметь 6 разделов / файлов.

См. Spark SQL документацию для получения дополнительной информации о параметрах режима.

Если вы хотите использовать свой собственный формат вывода, вы также сможете получить желаемое поведение с помощью RDD.

Посмотрите на следующие классы: FileOutputFormat , FileOutputCommitter

В формате выходного файла у вас есть метод с именем checkOutputSpecs, который проверяет, существует ли выходной каталог. В FileOutputCommitter у вас есть commitJob, который обычно переносит данные из временного каталога в свое конечное место.

Я еще не смог его проверить (сделаю это, как только у меня будет несколько бесплатных минут) но теоретически: если я расширяю FileOutputFormat и переопределяю checkOutputSpecs на метод, который не генерирует исключение в каталоге, уже существует, и отредактируйте метод commitJob моего пользовательского обработчика вывода для выполнения той логики, которую я хочу (например, переопределить некоторые из файлов, добавьте другие), чем я также смогу добиться желаемого поведения с помощью RDD.

Формат вывода передается в: saveAsNewAPIHadoopFile (который также называется методом saveAsTextFile, чтобы фактически сохранить файлы). И коммиттер вывода настроен на уровне приложения.

Эта перегруженная версия функции сохранения работает для меня:

yourDF.save (outputPath, org.apache.spark.sql.SaveMode.valueOf («Overwrite»))

Приведенный выше пример заменит существующую папку. Сабемод также может принимать эти параметры ( https://spark.apache.org/docs/1.4.0/api/java/org/apache/spark/sql/SaveMode.html ):

Append: Режим добавления означает, что при сохранении DataFrame в источнике данных, если данные / таблица уже существует, ожидается, что содержимое DataFrame будет добавлено к существующим данным.

ErrorIfExists: Режим ErrorIfExists означает, что при сохранении DataFrame в источнике данных, если данные уже существуют, ожидается, что будет выбрано исключение.

Игнорировать: режим Ignore означает, что при сохранении DataFrame в источнике данных, если данные уже существует, ожидается, что операция сохранения не сохранит содержимое DataFrame и не изменит существующие данные.

  val jobName = "WordCount";
  //overwrite the output directory in spark  set("spark.hadoop.validateOutputSpecs", "false")
  val conf = new 
  SparkConf().setAppName(jobName).set("spark.hadoop.validateOutputSpecs", "false");
  val sc = new SparkContext(conf)