Enregistrer le contenu de Spark DataFrame en tant que fichier CSV unique

Dire que j'ai une Étincelle DataFrame qui je veux enregistrer en tant que fichier CSV. Après Étincelle 2.0.0 , DataFrameWriter classe prend en charge l'enregistrement en tant que fichier CSV.

Le comportement par défaut est d'enregistrer la sortie dans plusieurs partie-*.csv les fichiers dans le chemin d'accès fourni.

Comment puis-je enregistrer un DF avec :

Chemin de la cartographie pour le nom exact du fichier au lieu de dossier de
En-tête disponible en première ligne
Enregistrer en tant que fichier unique au lieu de plusieurs fichiers.

Un moyen de traiter avec elle, est d'unir les DF et puis enregistrez le fichier.

df.coalesce(1).write.option("header", "true").csv("sample_file.csv")

Cependant, cela a le désavantage de collecte en Maître de la machine et doit avoir un maître avec suffisamment de mémoire.

Est-il possible de n'écrire qu'un seul fichier CSV sans l'aide de fusionnent ? Si non, est-il un moyen efficace que le code ci-dessus ?

source d'informationauteur Spandan Brahmbhatt | 2017-01-31

2

Utilisation:
df.toPandas().to_csv("sample_file.csv", header=True)

Voir la documentation pour plus de détails:
https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=dataframe#pyspark.sql.DataFrame.toPandas
1

Cette solution est basée sur un Script Shell qui n'est pas parallélisé, mais est toujours très rapide, en particulier sur les Ssd. Il utilise cat et redirection de la sortie sur les systèmes Unix. Supposons que le CSV dans le répertoire contenant les partitions est situé sur /my/csv/dir et que le fichier de sortie est /my/csv/output.csv:
```
#!/bin/bash
echo "col1,col2,col3" > /my/csv/output.csv
for i in /my/csv/dir/*.csv ; do
    echo "Processing $i"
    cat $i >> /my/csv/output.csv
    rm $i
done
echo "Done"
```
Il va supprimer chaque partition après l'ajoutant à la finale CSV afin de libérer de l'espace.

"col1,col2,col3" est le CSV en-tête (ici, nous avons trois colonnes de nom col1col2 et col3). Vous devez dire Étincelle pour ne pas mettre la tête dans chaque partition (ce qui est accompli avec .option("header", "false") parce que le Script Shell va le faire.

Pour ceux qui veulent pour ce faire, voici comment je me suis fait à l'aide de l'étincelle 2.1 en scala, avec quelques java.nio.file aider.

Basé sur https://fullstackml.com/how-to-export-data-frame-from-apache-spark-3215274ee9d6

    val df: org.apache.spark.sql.DataFrame = ??? //data frame to write
    val file: java.nio.file.Path = ??? //target output file (i.e. 'out.csv')

    import scala.collection.JavaConversions._

    //write csv into temp directory which contains the additional spark output files
    //could use Files.createTempDirectory instead
    val tempDir = file.getParent.resolve(file.getFileName + "_tmp")
    df.coalesce(1)
        .write.format("com.databricks.spark.csv")
        .option("header", "true")
        .save(tempDir.toAbsolutePath.toString)

    //find the actual csv file
    val tmpCsvFile = Files.walk(tempDir, 1).iterator().toSeq.find { p => 
        val fname = p.getFileName.toString
        fname.startsWith("part-00000") && fname.endsWith(".csv") && Files.isRegularFile(p)
    }.get

    //move to desired final path
    Files.move(tmpCsvFile, file)

    //delete temp directory
    Files.walk(tempDir)
        .sorted(java.util.Comparator.reverseOrder())
        .iterator().toSeq
        .foreach(Files.delete(_))

1

C'est la façon dont l'informatique distribuée travail! Plusieurs fichiers à l'intérieur d'un répertoire est exactement la façon dont l'informatique distribuée œuvres, ce n'est pas un problème, tout comme tous les logiciels peuvent le manipuler.

Votre question devrait être "comment est-il possible de télécharger un fichier CSV composé de plusieurs fichiers?" -> il y a déjà lof de solutions.

Une autre approche pourrait être d'utiliser une Étincelle comme une source JDBC (avec le génial Étincelle d'Aubaines serveur), écrire une requête SQL et de transformer le résultat au format CSV.

Afin de prévenir OOM dans le pilote (depuis le pilote obtiendrez TOUS les
les données), utilise les recueillir
(spark.sql.thriftServer.incrementalCollect=true), plus d'infos à
http://www.russellspitzer.com/2017/05/19/Spark-Sql-Thriftserver/.

Petit récapitulatif sur Spark "partition de données" le concept:

INPUT (X PARTITIONs) -> COMPUTING (Y PARTITIONs) -> OUTPUT (Z PARTITIONs)

Entre les "étapes", les données peuvent être transférées entre des partitions, c'est le "shuffle". Vous voulez "Z" = 1, mais avec Y > 1, sans shuffle? cela est impossible.
1

Viens de résoudre moi-même à l'aide de pyspark avec dbutils pour obtenir le .csv et renommer le nom de fichier voulu.
```
save_location= "s3a://landing-bucket-test/export/"+year
csv_location = save_location+"temp.folder'
file_location = save_location+'export.csv'

df.repartition(1).write.csv(path=csv_location, mode="append", header="true")

file = dbutils.fs.ls(csv_location)[-1].path
dbutils.fs.cp(file, file_location)
dbutils.fs.rm(csv_location, recurse=True)
```
Cette réponse peut être améliorée en n'utilisant pas de [-1], mais le .csv semble toujours être le dernier dans le dossier. Solution Simple et rapide si vous ne travaillez sur des fichiers plus petits et peuvent utiliser répartition(1) ou fusionner(1).

La FileUtil.copyMerge() de l'Hadoop API devrait résoudre votre problème.

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs._

def merge(srcPath: String, dstPath: String): Unit =  {
   val hadoopConfig = new Configuration()
   val hdfs = FileSystem.get(hadoopConfig)
   FileUtil.copyMerge(hdfs, new Path(srcPath), hdfs, new Path(dstPath), true, hadoopConfig, null) 
   //the "true" setting deletes the source files once they are merged into the new output
}

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

df.coalesce(1).write.option("inferSchema","true").csv("/newFolder",header = 
'true',dateFormat = "yyyy-MM-dd HH:mm:ss")

Vous devez vous connecter pour publier un commentaire.