Jak odczytać wiele plików tekstowych w jednym RDD?

Chcę odczytać kilka plików tekstowych z lokalizacji hdfs i wykonać mapowanie w iteracji za pomocą Spark.

JavaRDD<String> records = ctx.textFile(args[1], 1); jest w stanie odczytać tylko jeden plik na raz.

Chcę odczytać więcej niż jeden plik i przetworzyć je jako pojedynczy RDD. W jaki sposób?

Możesz określić całe katalogi, używać symboli wieloznacznych, a nawet CSV katalogów i symboli wieloznacznych. Na przykład:

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

Jak zauważa Nick Chammas, jest to ujawnienie Hadoopa FileInputFormati dlatego działa również z Hadoop (i Scalding).

Użyj unionw następujący sposób:

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

Następnie bigRddjest RDD ze wszystkimi plikami.

Możesz użyć pojedynczego wywołania textFile, aby odczytać wiele plików. Scala:

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

Możesz tego użyć

Najpierw możesz zdobyć bufor / listę ścieżek 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
  }

Teraz przekaż ten obiekt List do następującego fragmentu kodu, uwaga: sc jest obiektem SQLContext

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

Teraz masz ostateczną Unified RDD, tj. Df

Opcjonalnie, a także można podzielić na partycje w jednym BigRDD

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

Ponowne partycjonowanie zawsze działa: D

W PySpark znalazłem dodatkowy użyteczny sposób na parsowanie plików. Być może w Scali jest odpowiednik, ale nie czuję się wystarczająco dobrze wymyślając tłumaczenie robocze. W rzeczywistości jest to wywołanie textFile z dodanymi etykietami (w poniższym przykładzie klucz = nazwa pliku, wartość = 1 linia z pliku).

„Oznaczony” plik tekstowy

Wejście:

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)

wyjście: tablica z każdym wpisem zawierającym krotkę przy użyciu nazwa-pliku-klucza i wartość = każdy wiersz pliku. (Technicznie, korzystając z tej metody, możesz również użyć innego klucza oprócz nazwy ścieżki do pliku - być może reprezentacji mieszającej, aby zaoszczędzić na pamięci). to znaczy.

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

Możesz także połączyć ponownie jako listę linii:

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

Lub połącz ponownie całe pliki z powrotem w pojedyncze ciągi znaków (w tym przykładzie wynik jest taki sam, jak wynik z całych plików tekstowych, ale z ciągiem „file:” usuniętym ze ścieżki pliku).

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

możesz użyć

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

tutaj otrzymasz ścieżkę do pliku i zawartość tego pliku. dzięki czemu można wykonać dowolną akcję całego pliku w tym samym czasie, co oszczędza koszty ogólne

Wszystkie odpowiedzi są poprawne z sc.textFile

Zastanawiałem się, dlaczego nie wholeTextFilesNa przykład w tym przypadku ...

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

jednym ograniczeniem jest to, że musimy ładować małe pliki, w przeciwnym razie wydajność będzie niska i może prowadzić do OOM.

Uwaga :

• Pełny plik powinien pasować do pamięci
• Dobre dla formatów plików, których NIE można podzielić według linii ... takich jak pliki XML

Dalsze informacje do odwiedzenia

Dostępne jest proste, czyste rozwiązanie. Użyj metody wholeTextFiles (). Spowoduje to pobranie katalogu i utworzenie pary klucz-wartość. Zwrócony RDD będzie parą RDD. Znajdź poniżej opis z dokumentacji Spark :

SparkContext.wholeTextFiles pozwala czytać katalog zawierający wiele małych plików tekstowych i zwraca każdy z nich jako pary (nazwa pliku, treść). Jest to w przeciwieństwie do textFile, który zwraca jeden rekord na linię w każdym pliku

WYPRÓBUJ TEN interfejs używany do zapisywania DataFrame w zewnętrznych systemach pamięci masowej (np. Systemach plików, sklepach z kluczowymi wartościami itp.). Użyj DataFrame.write (), aby uzyskać do tego dostęp.

Nowości w wersji 1.4.

csv (ścieżka, tryb = Brak, kompresja = Brak, sep = Brak, cytat = Brak, escape = Brak, nagłówek = Brak, nullValue = Brak, escapeQuotes = Brak, quoteAll = Brak, dateFormat = Brak, timestampFormat = Brak) Zapisuje zawartość DataFrame w formacie CSV pod określoną ścieżką.

Parametry: ścieżka - ścieżka w dowolnym trybie systemu plików obsługiwanym przez Hadoop - określa zachowanie operacji składowania, gdy dane już istnieją.

append: Dołącz zawartość DataFrame do istniejących danych. nadpisz: nadpisz istniejące dane. ignore: cicho zignoruj ​​tę operację, jeśli dane już istnieją. błąd (przypadek domyślny): Zgłaszaj wyjątek, jeśli dane już istnieją. kompresja - kodek kompresji używany podczas zapisywania do pliku. Może to być jedna ze znanych krótkich nazw bez rozróżniania wielkości liter (none, bzip2, gzip, lz4, snappy i deflate). sep - ustawia pojedynczy znak jako separator dla każdego pola i wartości. Jeśli ustawiono Brak, używa wartości domyślnej,,. quote - ustawia pojedynczy znak używany do zmiany wartości cytowanych, w których separator może być częścią wartości. Jeśli ustawiony jest None, używa wartości domyślnej, „. Jeśli chcesz wyłączyć cytaty, musisz ustawić pusty ciąg. Escape - ustawia pojedynczy znak używany do zmiany znaczenia cytatów wewnątrz już cytowanej wartości. Jeśli None jest ustawiony , używa wartości domyślnej, \ escapeQuotes - Flaga wskazująca, czy wartości zawierające cudzysłowy zawsze powinny być ujęte w cudzysłowy. Jeśli ustawiono Brak, używa domyślnej wartości true, unikając wszystkich wartości zawierających znak cudzysłowu. quoteAll - Flaga wskazująca, czy wszystkie wartości powinny być zawsze ujęte w cudzysłów. Jeśli ustawiono Brak, używa wartości domyślnej false, a jedynie wartości specjalne zawierające znak cudzysłowu. nagłówek - zapisuje nazwy kolumn jako pierwszy wiersz. Jeśli ustawiono None, używa wartości domyślnej, false. nullValue - ustawia ciąg reprezentujący wartość zerową. Jeśli ustawiono Brak, używa wartości domyślnej, pusty ciąg. dateFormat - ustawia ciąg wskazujący format daty. Niestandardowe formaty dat są zgodne z formatami java.text.SimpleDateFormat. Dotyczy to typu daty. Jeśli ustawiono Brak, używa wartości domyślnej rrrr-MM-dd. timestampFormat - ustawia ciąg znaków wskazujący format znacznika czasu. Niestandardowe formaty dat są zgodne z formatami java.text.SimpleDateFormat. Dotyczy to typu znacznika czasu. Jeśli ustawiony jest None, używa wartości domyślnej rrrr-MM-dd'T'HH: mm: ss.SSSZZ.

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