Spark Contexte fichier Texte: de charger plusieurs fichiers

J'ai besoin de traiter plusieurs fichiers dispersés sur différents annuaires. Je voudrais charger tous ces en un seul RDD et ensuite effectuer map/reduce sur elle. Je vois que SparkContext est capable de charger plusieurs fichiers à partir d'un seul répertoire à l'aide de caractères génériques. Je ne suis pas sûr de la façon de charger des fichiers à partir de plusieurs dossiers.

L'extrait de code suivant échoue:

for fileEntry in files:
    fileName = basePath + "/" + fileEntry
    lines = sc.textFile(fileName)
    if retval == None:
        retval = lines
    else:
        retval = sc.union(retval, lines)

Cela échoue sur la troisième boucle avec le message d'erreur suivant:

retval = sc.union(retval, lines)
TypeError: union() takes exactly 2 arguments (3 given)

Qui est bizarre étant donné que je suis en fournissant seulement 2 arguments. Tous les pointeurs apprécié.

  • ..mais le premier argument est self. À partir de la docs, vous avez besoin sc.union([retval,lines])
  • Laissez-moi essayer. Je suis surpris que ce serait travailler pour 2 boucles et d'échouer sur la troisième ...
  • Qui a fait le tour. Merci Jonathan!
  • Je viens de réaliser que vous pouvez utiliser sc.textFile(','.join(files)) de la lire en un seul coup.
InformationsquelleAutor Raj | 2014-04-30
  1. 43

    Comment à ce sujet phrasé à la place?

    sc.union([sc.textFile(basepath + "/" + f) for f in files])

    En Scala SparkContext.union() a deux variantes, l'une qui prend vararg arguments, et celui qui prend une liste. Seule la seconde existe en Python (depuis le Python n'a pas de polymorphisme).

    Mise à JOUR

    Vous pouvez utiliser un seul textFile appel à lire plusieurs fichiers.

    sc.textFile(','.join(files))
    • Merci Daniel. Ma question est peut-être Python centrique. Votre extrait semble Scala,
    • Ah, pourquoi n'ai-je pas réaliser que?! Il n'existe pas de fonction de polymorphisme en Python, de sorte qu'une seule forme de SparkContext.union() peuvent être exposés. Ils ont choisi d'exposer celui qui prend une liste, et non pas celui de prendre un vararg. (Comme Jonathan dit.)
    • J'ai corrigé la réponse à Python à la place de la Scala.
    • Il y a une faute de frappe dans cette réponse, mais je ne peux pas le modifier car il n'est pas 6 caractères: La "\n" devrait être un "/".
    • Merci! Étrange erreur d'en faire... \n et / ne sont nulle part près de l'autre sur le clavier :).
    InformationsquelleAutor Daniel Darabos
  2. 13

    Je résoudre des problèmes en utilisant des caractères génériques.

    par exemple, j'ai trouvé certains caractères dans les fichiers que je veux charger dans spark,

    dir

    subdir1/folder1/x.txt

    subdir2/folder2/y.txt

    vous pouvez utiliser la phrase suivante 

    sc.textFile("dir/*/*/*.txt")

    pour charger tous les fichiers relatifs.

    Le caractère générique '*' ne fonctionne que dans un seul répertoire de niveau， qui n'est pas récursive.

    InformationsquelleAutor fibonacci
  3. 2

    Vous pouvez utiliser la fonction suivante de SparkContext:

    wholeTextFiles(path: String, minPartitions: Int = defaultMinPartitions): RDD[(String, String)]

    Lire un répertoire de fichiers texte à partir de HDFS, un système de fichiers local (disponible sur tous les nœuds), ou tout Hadoop de fichier pris en charge système d'URI. Chaque fichier est lu comme un seul enregistrement et retourné dans une paire clé-valeur, où la clé est le chemin d'accès de chaque fichier, la valeur est le contenu de chaque fichier.

    https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.SparkContext

    • Cela fonctionne bien dans la plupart des cas, mais dans mon expérience, cela ne fonctionne pas lorsque la taille des fichiers est importante.
    InformationsquelleAutor Neil
  4. 1

    Vous pouvez utiliser cette

    D'abord, Vous pouvez obtenir un Tampon/Liste de S3 Chemins : 

    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
  }

    Maintenant Passer à cette Liste d'objet pour le morceau de code suivant, note : 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 une dernière Unifiée CA c'est à dire df

    En option, Et Vous pouvez également repartitionner en un seul BigRDD 

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

    Repartitionnement fonctionne toujours 😀

    InformationsquelleAutor Murtaza Kanchwala