Écriture d'un seul fichier CSV à l'aide de l'étincelle-csv

C'est la création d'un dossier avec plusieurs fichiers, parce que chaque partition est enregistré individuellement. Si vous avez besoin d'un fichier de sortie unique (dans un dossier), vous pouvez repartition (de préférence si en amont des données est grande, mais nécessite une lecture aléatoire):

df
   .repartition(1)
   .write.format("com.databricks.spark.csv")
   .option("header", "true")
   .save("mydata.csv")

ou coalesce:

df
   .coalesce(1)
   .write.format("com.databricks.spark.csv")
   .option("header", "true")
   .save("mydata.csv")

bloc de données avant de les enregistrer:

Toutes les données seront écrites dans mydata.csv/part-00000. Avant d'utiliser cette option assurez-vous de comprendre ce qui se passe et quel est le coût de transfert de toutes les données à un seul travailleur. Si vous utilisez le système de fichiers distribués avec la réplication, les données seront transférées à de nombreuses reprises, d'abord extraites d'un seul travailleur et par la suite répartis sur les nœuds de stockage.

Alternativement, vous pouvez laisser votre code comme il est et l'utilisation à des fins générales des outils comme cat ou HDFS getmerge tout simplement de fusionner toutes les pièces par la suite.

Je suis peut-être un peu tard pour le jeu ici, mais à l'aide de coalesce(1) ou repartition(1) peut travailler pour de petits ensembles de données, mais de grands ensembles de données seront tous jetés dans une partition sur un nœud. C'est probablement à jeter OOM erreurs, ou, au mieux, processus lentement.

Je vous suggère fortement que vous utilisez le FileUtil.copyMerge() fonction de l'Hadoop API. Cela permettra de fusionner les sorties dans un fichier unique.

MODIFIER - Ce effectivement regroupe les données pour le conducteur plutôt que d'un exécuteur testamentaire nœud. Coalesce() serait bien si un seul exécuteur testamentaire a plus de RAM pour que le pilote.

EDIT 2: copyMerge() est retiré dans Hadoop 3.0. Voir la suite de débordement de pile dans l'article pour plus d'informations sur la façon de travailler avec la version la plus récente: Hadoop comment faire CopyMerge dans Hadoop 3.0

repartitionner/fusionner pour 1 partition avant de vous enregistrer (si vous souhaitez toujours obtenir un dossier, mais il aurait une partie de fichier)

Une solution qui fonctionne pour S3 modifié à partir de Minkymorgan.

Tout simplement passer le temporaire partitionné chemin d'accès au répertoire (avec un nom différent de chemin final) comme le srcPath et seule finale d'un fichier csv/txt comme destPath Spécifier également deleteSource si vous souhaitez supprimer le répertoire d'origine.

/**
* Merges multiple partitions of spark text file output into single file. 
* @param srcPath source directory of partitioned files
* @param dstPath output path of individual path
* @param deleteSource whether or not to delete source directory after merging
* @param spark sparkSession
*/
def mergeTextFiles(srcPath: String, dstPath: String, deleteSource: Boolean): Unit =  {
  import org.apache.hadoop.fs.FileUtil
  import java.net.URI
  val config = spark.sparkContext.hadoopConfiguration
  val fs: FileSystem = FileSystem.get(new URI(srcPath), config)
  FileUtil.copyMerge(
    fs, new Path(srcPath), fs, new Path(dstPath), deleteSource, config, null
  )
}

Il y a un moyen de plus pour utiliser Java

import java.io._

def printToFile(f: java.io.File)(op: java.io.PrintWriter => Unit) 
  {
     val p = new java.io.PrintWriter(f);  
     try { op(p) } 
     finally { p.close() }
  } 

printToFile(new File("C:/TEMP/df.csv")) { p => df.collect().foreach(p.println)}