Como ler vários arquivos de texto em um único RDD?

Eu quero ler um monte de arquivos de texto em um local hdfs e executar o mapeamento nele em uma iteração usando o spark.

JavaRDD<String> records = ctx.textFile(args[1], 1); é capaz de ler apenas um arquivo por vez.

Quero ler mais de um arquivo e processá-los como um único RDD. Quão?

Você pode especificar diretórios inteiros, usar curingas e até CSV de diretórios e curingas. Por exemplo:

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

Como Nick Chammas aponta, essa é uma exposição do Hadoop FileInputFormate, portanto, isso também funciona com o Hadoop (e o Scalding).

Use da unionseguinte maneira:

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

Então o bigRddé o RDD com todos os arquivos.

Você pode usar uma única chamada textFile para ler vários arquivos. Scala:

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

Você pode usar isso

Primeiro, você pode obter um buffer / lista de caminhos 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
  }

Agora passe este objeto List para o seguinte trecho de código, observe: sc é um objeto do SQLContext

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

Agora você tem um RDD Unificado final, isto é, df

Opcional E você também pode reparticioná-lo em um único BigRDD

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

O reparticionamento sempre funciona: D

No PySpark, encontrei uma maneira útil adicional de analisar arquivos. Talvez haja um equivalente em Scala, mas não me sinto confortável o suficiente com uma tradução de trabalho. Com efeito, é uma chamada textFile com a adição de rótulos (no exemplo abaixo, a chave = nome do arquivo, valor = 1 linha do arquivo).

TextFile "Rotulado"

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)

output: array com cada entrada contendo uma tupla usando o nome do arquivo como chave e com value = cada linha do arquivo. (Tecnicamente, usando esse método, você também pode usar uma chave diferente, além do nome real do caminho do arquivo - talvez uma representação de hash para economizar memória). ie

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

Você também pode recombinar como uma lista de linhas:

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 recombine arquivos inteiros de volta para seqüências de caracteres únicas (neste exemplo, o resultado é o mesmo que você obtém de wholeTextFiles, mas com a sequência "file:" removida do caminho do arquivo.):

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

você pode usar

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

aqui você obterá o caminho do seu arquivo e o conteúdo desse arquivo. para que você possa executar qualquer ação de um arquivo inteiro de cada vez que salve a sobrecarga

Todas as respostas estão corretas com sc.textFile

Eu só estava me perguntando por que não wholeTextFilesPor exemplo, neste caso ...

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

Uma limitação é que, precisamos carregar arquivos pequenos, caso contrário, o desempenho será ruim e poderá levar ao OOM.

Nota :

• O wholefile deve caber na memória
• Bom para formatos de arquivo que NÃO são divisíveis por linha ... como arquivos XML

Referência adicional a visitar

Existe uma solução limpa direta disponível. Use o método wholeTextFiles (). Isso exigirá um diretório e formará um par de valores-chave. O RDD retornado será um par RDD. Encontre abaixo a descrição dos documentos do Spark :

SparkContext.wholeTextFiles permite ler um diretório que contém vários arquivos de texto pequenos e retorna cada um deles como pares (nome do arquivo, conteúdo). Isso contrasta com o textFile, que retornaria um registro por linha em cada arquivo

Experimente esta interface usada para gravar um DataFrame em sistemas de armazenamento externo (por exemplo, sistemas de arquivos, armazenamento de valores-chave, etc.). Use DataFrame.write () para acessar isso.

Novo na versão 1.4.

csv (caminho, modo = nenhum, compactação = nenhum, sep = nenhum, citação = nenhum, escape = nenhum, cabeçalho = nenhum, nullValue = nenhum, escapeQuotes = nenhum, quoteAll = nenhum, dateFormat = nenhum, timestampFormat = nenhum) Salva o conteúdo do DataFrame no formato CSV no caminho especificado.

Parâmetros: path - o caminho em qualquer modo de sistema de arquivos suportado pelo Hadoop - especifica o comportamento da operação de salvamento quando os dados já existem.

anexar: anexa o conteúdo deste DataFrame aos dados existentes. substituir: sobrescreve os dados existentes. ignorar: ignore silenciosamente esta operação se os dados já existirem. erro (caso padrão): lança uma exceção se os dados já existirem. compressão - codec de compressão a ser usado ao salvar em arquivo. Esse pode ser um dos nomes abreviados que não diferenciam maiúsculas de minúsculas (nenhum, bzip2, gzip, lz4, snappy e deflate). sep - define o caractere único como um separador para cada campo e valor. Se Nenhum estiver definido, ele usará o valor padrão,,. quote - define o caractere único usado para escapar dos valores entre aspas, onde o separador pode fazer parte do valor. Se Nenhum estiver definido, ele usará o valor padrão ". Se você desejar desativar as cotações, precisará definir uma sequência vazia. Escape - define o caractere único usado para escapar aspas dentro de um valor já citado. Se Nenhum estiver definido , ele usa o valor padrão, \ escapeQuotes - Um sinalizador que indica se os valores que contêm aspas sempre devem ser colocados entre aspas. Se Nenhum estiver definido, ele usará o valor padrão true, escapando a todos os valores que contêm um caractere de citação. quoteAll - Um sinalizador que indica se todos os valores devem sempre estar entre aspas. Se Nenhum estiver definido, ele usará o valor padrão false, somente valores de escape contendo um caractere de citação. header - escreve os nomes das colunas como a primeira linha. Se Nenhum estiver definido, ele usará o valor padrão, false. nullValue - define a representação de string de um valor nulo. Se Nenhum estiver definido, ele usará o valor padrão, sequência vazia. dateFormat - define a string que indica um formato de data. Os formatos de data personalizados seguem os formatos em java.text.SimpleDateFormat. Isso se aplica ao tipo de data. Se Nenhum estiver definido, ele usará o valor padrão, aaaa-MM-dd. timestampFormat - define a sequência que indica um formato de carimbo de data / hora. Os formatos de data personalizados seguem os formatos em java.text.SimpleDateFormat. Isso se aplica ao tipo de carimbo de data e hora. Se Nenhum estiver definido, ele usará o valor padrão, aaaa-MM-dd'T'HH: mm: ss.SSSZZ.

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