Comment puis-je faire (Spark1.6) saveAsTextFile pour ajouter le fichier existant?

Dans SparkSQL,j'utilise DF.wirte.mode(SaveMode.Ajouter).json(xxxx),mais cette méthode d'obtenir ces fichiers commeComment puis-je faire (Spark1.6) saveAsTextFile pour ajouter le fichier existant?

le nom de fichier est trop complexe et aléatoire,je ne peux pas utiliser l'api pour obtenir.Je tiens donc à utiliser saveAstextfile ,car le nom de fichier n'est pas complexe et régulière, mais je ne sais pas comment faire pour ajouter un fichier dans le même diretory?Apprécier votre temps.

avez-vous fixe elle même, je suis confronté à un même problème @yixiyix

OriginalL'auteur yjxyjx | 2016-05-04

  1. 2

    Vous pouvez essayer cette méthode que je trouve quelque part.
    Processus de Spark Streaming rdd et de les stocker à un seul fichier HDFS

        import org.apache.hadoop.fs.{ FileSystem, FileUtil, Path }

def saveAsTextFileAndMerge[T](hdfsServer: String, fileName: String, rdd: RDD[T]) = {
  val sourceFile = hdfsServer + "/tmp/"
  rdd.saveAsTextFile(sourceFile)
  val dstPath = hdfsServer + "/final/"
  merge(sourceFile, dstPath, fileName)
}

def merge(srcPath: String, dstPath: String, fileName: String): Unit = {
  val hadoopConfig = new Configuration()
  val hdfs = FileSystem.get(hadoopConfig)
  val destinationPath = new Path(dstPath)
  if (!hdfs.exists(destinationPath)) {
    hdfs.mkdirs(destinationPath)
  }
  FileUtil.copyMerge(hdfs, new Path(srcPath), hdfs, new Path(dstPath + "/" + fileName), false, hadoopConfig, null)
}
    Je crois que vous avez oublié d'importer : "import org.apache.hadoop.conf.Configuration"

    OriginalL'auteur Matiji66

  2. 2

    travaillé sur la Spark 1.5 , je pense que c'est le droit d'utilisation..

    dataframe.write().mode(SaveMode.Append).format(FILE_FORMAT).**partitionBy**("parameter1", "parameter2").save(path);

    OriginalL'auteur Ram Ghadiyaram

  3. 1

    Comme étincelle utilise HDFS, c'est la sortie standard qu'il produit. Vous pouvez utiliser le FileUtil fusionner les fichiers en un seul retour. C'est une solution efficace car il ne nécessite pas d'étincelle à recueillir l'ensemble des données en mémoire par le partitionnement en 1. C'est la démarche que j'ai suivi. 

    import org.apache.hadoop.fs.{FileSystem, FileUtil, Path}   

val hadoopConf = sqlContext.sparkContext.hadoopConfiguration
val hdfs = FileSystem.get(hadoopConf)
val mergedPath = "merged-" + filePath + ".json"
val merged = new Path(mergedPath)
if (hdfs.exists(merged)) {
  hdfs.delete(merged, true)
}
df.wirte.mode(SaveMode.Append).json(filePath)

FileUtil.copyMerge(hdfs, path, hdfs, merged, false, hadoopConf, null)

    Vous pouvez lire le même fichier à l'aide mergedPath emplacement. Espérons que cela aide.

    Merci, et je veux réaliser que ,par exemple, Dans HDFS, j'ai 3 fichiers comme partie-00000,partie-00001,partie-00002,ma demande est de faire de ces fichiers se transformer en partie-00000,je peux Donc utiliser copyMerge en existait fichier?
    Je ne suis pas très clair au sujet de votre question. Si vous vous demandez si vous pouvez fusionner la partie-00000,partie-00001,partie-00002 en partie-00000,, c'est ce que le code ci-dessus ne. Vous avez juste besoin de formuler la mergedPath comme vous le souhaitez. Est-ce que vous cherchez?
    Il est le coalesce fonction Étincelle pour tout rassembler dans un seul fichier.

    OriginalL'auteur NehaM