¿Cómo leer múltiples archivos de texto en un solo RDD?

Quiero leer un montón de archivos de texto desde una ubicación hdfs y realizar un mapeo en él en una iteración usando spark.

JavaRDD<String> records = ctx.textFile(args[1], 1); es capaz de leer solo un archivo a la vez.

Quiero leer más de un archivo y procesarlos como un único RDD. ¿Cómo?

Puede especificar directorios completos, usar comodines e incluso CSV de directorios y comodines. P.ej:

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

Como Nick Chammas señala, esta es una exposición de Hadoop FileInputFormaty, por lo tanto, esto también funciona con Hadoop (y Scalding).

Use de la unionsiguiente manera:

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

Entonces el bigRddes el RDD con todos los archivos.

Puede usar una sola llamada textFile para leer varios archivos. Scala:

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

Puedes usar esto

Primero puede obtener un Buffer / Lista de rutas 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
  }

Ahora pase este objeto List al siguiente fragmento de código, tenga en cuenta: sc es un objeto de SQLContext

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

Ahora tienes un RDD unificado final, es decir, df

Opcional, y también puede repartirlo en un solo BigRDD

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

Reparticionar siempre funciona: D

En PySpark, he encontrado una forma útil adicional de analizar archivos. Quizás haya un equivalente en Scala, pero no me siento lo suficientemente cómodo como para encontrar una traducción que funcione. Es, en efecto, una llamada textFile con la adición de etiquetas (en el ejemplo a continuación, la clave = nombre de archivo, valor = 1 línea del archivo).

Archivo de texto "etiquetado"

entrada:

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)

salida: matriz con cada entrada que contiene una tupla usando filename-as-key y con value = cada línea de archivo. (Técnicamente, con este método también puede usar una clave diferente además del nombre real de la ruta del archivo, tal vez una representación hash para guardar en la memoria). es decir.

[('/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'),
  ...]

También puede recombinarse como una lista de líneas:

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'])]

O recombine archivos completos de nuevo a cadenas individuales (en este ejemplo, el resultado es el mismo que el que obtiene de wholeTextFiles, pero con la cadena "file:" eliminada de la ruta de archivo):

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

puedes usar

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

aquí obtendrá la ruta de su archivo y el contenido de ese archivo. para que pueda realizar cualquier acción de un archivo completo a la vez que ahorre la sobrecarga

Todas las respuestas son correctas con sc.textFile

Me preguntaba por qué no. wholeTextFilesPor ejemplo, en este caso ...

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

Una limitación es que tenemos que cargar archivos pequeños, de lo contrario el rendimiento será malo y puede conducir a OOM.

Nota :

• Todo el archivo debe caber en la memoria
• Bueno para formatos de archivo que NO se pueden dividir por línea ... como archivos XML

Más referencia para visitar

Hay una solución limpia directa disponible. Use el método wholeTextFiles (). Esto tomará un directorio y formará un par de valores clave. El RDD devuelto será un par RDD. Encuentra debajo la descripción de Spark docs :

SparkContext.wholeTextFiles le permite leer un directorio que contiene múltiples archivos de texto pequeños y devuelve cada uno de ellos como pares (nombre de archivo, contenido). Esto está en contraste con textFile, que devolvería un registro por línea en cada archivo

PRUEBA ESTO interfaz utilizada para escribir un DataFrame en sistemas de almacenamiento externo (por ejemplo, sistemas de archivos, almacenes de valores clave, etc.). Use DataFrame.write () para acceder a esto.

Nuevo en la versión 1.4.

csv (ruta, modo = Ninguno, compresión = Ninguno, sep = Ninguno, comilla = Ninguno, escape = Ninguno, encabezado = Ninguno, nullValue = Ninguno, escapeQuotes = Ninguno, quoteAll = Ninguno, dateFormat = Ninguno, timestampFormat = Ninguno) Guarda el contenido del DataFrame en formato CSV en la ruta especificada.

Parámetros: ruta - la ruta en cualquier modo de sistema de archivos compatible con Hadoop - especifica el comportamiento de la operación de guardar cuando los datos ya existen.

append: añade contenido de este DataFrame a los datos existentes. sobrescribir: sobrescribe los datos existentes. ignorar: ignore silenciosamente esta operación si ya existen datos. error (caso predeterminado): arroje una excepción si los datos ya existen. compresión - códec de compresión para usar al guardar en un archivo. Este puede ser uno de los nombres abreviados que no distinguen entre mayúsculas y minúsculas (none, bzip2, gzip, lz4, snappy and deflate). sep: establece el carácter único como separador para cada campo y valor. Si Ninguno está configurado, usa el valor predeterminado,,. quote: establece el carácter único utilizado para escapar de los valores entre comillas donde el separador puede ser parte del valor. Si Ninguno está configurado, utiliza el valor predeterminado, ". Si desea desactivar las comillas, debe establecer una cadena vacía. Escape: establece el carácter único utilizado para escapar de las comillas dentro de un valor ya citado. Si Ninguno está configurado , utiliza el valor predeterminado, \ escapeQuotes: una marca que indica si los valores que contienen comillas siempre deben estar entre comillas. Si Ninguno está configurado, utiliza el valor predeterminado verdadero, escapando a todos los valores que contienen un carácter de comillas. quoteAll: una marca que indica si todos los valores siempre deben estar entre comillas. Si se establece Ninguno, utiliza el valor predeterminado falso, solo escapa los valores que contienen un carácter de comillas. encabezado: escribe los nombres de las columnas como la primera línea. Si Ninguno está configurado, utiliza el valor predeterminado, falso. nullValue: establece la representación de cadena de un valor nulo. Si Ninguno está configurado, utiliza el valor predeterminado, cadena vacía. dateFormat: establece la cadena que indica un formato de fecha. Los formatos de fecha personalizados siguen los formatos en java.text.SimpleDateFormat. Esto se aplica al tipo de fecha. Si Ninguno está configurado, utiliza el valor del valor predeterminado, aaaa-MM-dd. timestampFormat: establece la cadena que indica un formato de marca de tiempo. Los formatos de fecha personalizados siguen los formatos en java.text.SimpleDateFormat. Esto se aplica al tipo de marca de tiempo. Si se establece Ninguno, utiliza el valor del valor predeterminado, aaaa-MM-dd'T'HH: mm: ss.SSSZZ.

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