Comment lire plusieurs fichiers texte dans un seul RDD?

Je veux lire un tas de fichiers texte à partir d'un emplacement hdfs et effectuer un mappage dessus dans une itération à l'aide de spark.

JavaRDD<String> records = ctx.textFile(args[1], 1); est capable de lire un seul fichier à la fois.

Je veux lire plus d'un fichier et les traiter comme un seul RDD. Comment?

Vous pouvez spécifier des répertoires entiers, utiliser des caractères génériques et même des fichiers CSV de répertoires et des caractères génériques. Par exemple:

sc.textFile("/my/dir1,/my/paths/part-00[0-5]*,/another/dir,/a/specific/file")

Comme le souligne Nick Chammas, il s'agit d'une exposition de Hadoop FileInputFormatet donc cela fonctionne également avec Hadoop (et Scalding).

Utilisez unioncomme suit:

val sc = new SparkContext(...)
val r1 = sc.textFile("xxx1")
val r2 = sc.textFile("xxx2")
...
val rdds = Seq(r1, r2, ...)
val bigRdd = sc.union(rdds)

Ensuite, bigRddc'est le RDD avec tous les fichiers.

Vous pouvez utiliser un seul appel textFile pour lire plusieurs fichiers. Scala:

sc.textFile(','.join(files)) 

Vous pouvez utiliser ceci

Vous pouvez d'abord obtenir un tampon / une liste de chemins S3:

import scala.collection.JavaConverters._
import java.util.ArrayList
import com.amazonaws.services.s3.AmazonS3Client
import com.amazonaws.services.s3.model.ObjectListing
import com.amazonaws.services.s3.model.S3ObjectSummary
import com.amazonaws.services.s3.model.ListObjectsRequest

def listFiles(s3_bucket:String, base_prefix : String) = {
    var files = new ArrayList[String]

    //S3 Client and List Object Request
    var s3Client = new AmazonS3Client();
    var objectListing: ObjectListing = null;
    var listObjectsRequest = new ListObjectsRequest();

    //Your S3 Bucket
    listObjectsRequest.setBucketName(s3_bucket)

    //Your Folder path or Prefix
    listObjectsRequest.setPrefix(base_prefix)

    //Adding s3:// to the paths and adding to a list
    do {
      objectListing = s3Client.listObjects(listObjectsRequest);
      for (objectSummary <- objectListing.getObjectSummaries().asScala) {
        files.add("s3://" + s3_bucket + "/" + objectSummary.getKey());
      }
      listObjectsRequest.setMarker(objectListing.getNextMarker());
    } while (objectListing.isTruncated());

    //Removing Base Directory Name
    files.remove(0)

    //Creating a Scala List for same
    files.asScala
  }

Passez maintenant cet objet List au morceau de code suivant, notez: sc est un objet de SQLContext

var df: DataFrame = null;
  for (file <- files) {
    val fileDf= sc.textFile(file)
    if (df!= null) {
      df= df.unionAll(fileDf)
    } else {
      df= fileDf
    }
  }

Maintenant vous avez un final unifié RDD ie df

Facultatif, et vous pouvez également le repartitionner dans un seul BigRDD

val files = sc.textFile(filename, 1).repartition(1)

Le repartitionnement fonctionne toujours: D

Dans PySpark, j'ai trouvé un moyen utile supplémentaire pour analyser les fichiers. Il existe peut-être un équivalent dans Scala, mais je ne suis pas assez à l'aise pour proposer une traduction fonctionnelle. Il s'agit en fait d'un appel textFile avec l'ajout d'étiquettes (dans l'exemple ci-dessous, la clé = nom de fichier, valeur = 1 ligne du fichier).

TextFile "étiqueté"

contribution:

import glob
from pyspark import SparkContext
SparkContext.stop(sc)
sc = SparkContext("local","example") # if running locally
sqlContext = SQLContext(sc)

for filename in glob.glob(Data_File + "/*"):
    Spark_Full += sc.textFile(filename).keyBy(lambda x: filename)

sortie: tableau avec chaque entrée contenant un tuple utilisant filename-as-key et avec valeur = chaque ligne de fichier. (Techniquement, en utilisant cette méthode, vous pouvez également utiliser une clé différente en plus du nom de chemin de fichier réel - peut-être une représentation de hachage à enregistrer sur la mémoire). c'est à dire.

[('/home/folder_with_text_files/file1.txt', 'file1_contents_line1'),
 ('/home/folder_with_text_files/file1.txt', 'file1_contents_line2'),
 ('/home/folder_with_text_files/file1.txt', 'file1_contents_line3'),
 ('/home/folder_with_text_files/file2.txt', 'file2_contents_line1'),
  ...]

Vous pouvez également recombiner soit sous forme de liste de lignes:

Spark_Full.groupByKey().map(lambda x: (x[0], list(x[1]))).collect()

[('/home/folder_with_text_files/file1.txt', ['file1_contents_line1', 'file1_contents_line2','file1_contents_line3']),
 ('/home/folder_with_text_files/file2.txt', ['file2_contents_line1'])]

Ou recombinez des fichiers entiers en chaînes uniques (dans cet exemple, le résultat est le même que celui que vous obtenez de wholeTextFiles, mais avec la chaîne "file:" supprimée du chemin de fichier.):

Spark_Full.groupByKey().map(lambda x: (x[0], ' '.join(list(x[1])))).collect()

vous pouvez utiliser

JavaRDD<String , String> records = sc.wholeTextFiles("path of your directory")

ici, vous obtiendrez le chemin de votre fichier et le contenu de ce fichier. afin que vous puissiez effectuer n'importe quelle action d'un fichier entier à un moment qui économise la surcharge

Toutes les réponses sont correctes avec sc.textFile

Je me demandais juste pourquoi pas wholeTextFilesPar exemple, dans ce cas ...

val minPartitions = 2
val path = "/pathtohdfs"
    sc.wholeTextFiles(path,minPartitions)
      .flatMap{case (path, text) 
    ...

une limitation est que nous devons charger de petits fichiers sinon les performances seront mauvaises et peuvent conduire à un MOO.

Remarque :

• L'ensemble du fichier doit tenir dans la mémoire
• Idéal pour les formats de fichiers qui ne sont PAS divisibles par ligne ... tels que les fichiers XML

Autre référence à visiter

Il existe une solution simple et propre. Utilisez la méthode wholeTextFiles (). Cela prendra un répertoire et forme une paire clé / valeur. Le RDD retourné sera une paire RDD. Retrouvez ci-dessous la description tirée de la documentation Spark :

SparkContext.wholeTextFiles vous permet de lire un répertoire contenant plusieurs petits fichiers texte et renvoie chacun d'eux sous forme de paires (nom de fichier, contenu). Ceci est en contraste avec textFile, qui renverrait un enregistrement par ligne dans chaque fichier

ESSAYEZ CETTE Interface utilisée pour écrire un DataFrame sur des systèmes de stockage externes (par exemple, des systèmes de fichiers, des magasins clé-valeur, etc.). Utilisez DataFrame.write () pour y accéder.

Nouveau dans la version 1.4.

csv (chemin, mode = None, compression = None, sep = None, quote = None, escape = None, header = None, nullValue = None, escapeQuotes = None, quoteAll = None, dateFormat = None, timestampFormat = None) Enregistre le contenu du DataFrame au format CSV au chemin spécifié.

Paramètres: chemin - le chemin dans n'importe quel mode de système de fichiers pris en charge par Hadoop - spécifie le comportement de l'opération de sauvegarde lorsque des données existent déjà.

append: ajoute le contenu de ce DataFrame aux données existantes. écraser: écraser les données existantes. ignore: ignorez en silence cette opération si des données existent déjà. erreur (cas par défaut): lance une exception si des données existent déjà. compression - codec de compression à utiliser lors de l'enregistrement dans un fichier. Il peut s'agir de l'un des noms abrégés connus insensibles à la casse (none, bzip2, gzip, lz4, snappy et deflate). sep - définit le caractère unique comme séparateur pour chaque champ et valeur. Si Aucun est défini, il utilise la valeur par défaut,,. quote - définit le caractère unique utilisé pour échapper les valeurs entre guillemets où le séparateur peut faire partie de la valeur. Si Aucun est défini, il utilise la valeur par défaut, ". Si vous souhaitez désactiver les guillemets, vous devez définir une chaîne vide. Escape - définit le caractère unique utilisé pour échapper les guillemets dans une valeur déjà citée. Si Aucun est défini , il utilise la valeur par défaut, \ escapeQuotes - Un indicateur indiquant si les valeurs contenant des guillemets doivent toujours être placées entre guillemets. Si None est défini, il utilise la valeur par défaut true, en échappant à toutes les valeurs contenant un caractère guillemet. quoteAll - Un indicateur indiquant si toutes les valeurs doivent toujours être placées entre guillemets. Si None est défini, il utilise la valeur par défaut false, en échappant uniquement les valeurs contenant un caractère guillemet. header - écrit les noms des colonnes comme première ligne. Si None est défini, il utilise la valeur par défaut, false. nullValue - définit la représentation sous forme de chaîne d'une valeur nulle. Si None est défini, il utilise la valeur par défaut, une chaîne vide. dateFormat - définit la chaîne qui indique un format de date. Les formats de date personnalisés suivent les formats de java.text.SimpleDateFormat. Cela s'applique au type de date. Si Aucun est défini, il utilise la valeur par défaut, aaaa-MM-jj. timestampFormat - définit la chaîne qui indique un format d'horodatage. Les formats de date personnalisés suivent les formats de java.text.SimpleDateFormat. Cela s'applique au type d'horodatage. Si Aucun est défini, il utilise la valeur par défaut, aaaa-MM-jj'T'HH: mm: ss.SSSZZ.

rdd = textFile('/data/{1.txt,2.txt}')