Как прочитать несколько текстовых файлов в одном RDD?

Я хочу прочитать несколько текстовых файлов из местоположения hdfs и выполнить сопоставление с ним в итерации, используя spark.

JavaRDD<String> records = ctx.textFile(args[1], 1); способен читать только один файл за раз.

Я хочу прочитать более одного файла и обработать их как один RDD. Как?

Вы можете указать целые каталоги, использовать подстановочные знаки и даже CSV каталогов и подстановочные знаки. Например:

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

Как отмечает Ник Чаммас, это разоблачение Hadoop FileInputFormatи, следовательно, это также работает с Hadoop (и Scalding).

Используйте unionследующим образом:

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

Тогда bigRddэто СДР со всеми файлами.

Вы можете использовать один вызов textFile для чтения нескольких файлов. Scala:

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

Вы можете использовать это

Сначала вы можете получить буфер / список путей 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
  }

Теперь передайте этот объект List следующему коду, обратите внимание: sc является объектом SQLContext

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

Теперь вы получили окончательный Унифицированный СДР, т.е. DF

Необязательно, и вы также можете перераспределить его в одном BigRDD

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

Перераспределение всегда работает: D

В PySpark я нашел еще один полезный способ разбора файлов. Возможно, в Scala есть эквивалент, но мне не совсем удобно придумывать рабочий перевод. По сути, это вызов textFile с добавлением меток (в приведенном ниже примере ключ = имя файла, значение = 1 строка из файла).

«Помеченный» textFile

вход:

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: массив с каждой записью, содержащей кортеж, используя filename-as-key и со значением = каждая строка файла. (Технически, используя этот метод, вы также можете использовать другой ключ помимо фактического имени пути к файлу - возможно, хеш-представление для экономии памяти). то есть.

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

Вы также можете рекомбинировать в виде списка строк:

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

Или рекомбинируйте целые файлы обратно в одну строку (в этом примере результат совпадает с тем, что вы получаете от wholeTextFiles, но со строкой «file:», удаленной из пути к файлу.):

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

ты можешь использовать

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

здесь вы получите путь к вашему файлу и его содержимое. так что вы можете выполнять любое действие целого файла за раз, что экономит накладные расходы

Все ответы верны с sc.textFile

Мне просто интересно, почему нет wholeTextFilesНапример, в этом случае ...

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

Одним из ограничений является то, что мы должны загружать небольшие файлы, иначе производительность будет плохой и может привести к OOM.

Примечание :

• Весь файл должен уместиться в памяти
• Подходит для форматов файлов, которые НЕ разделяются по строкам ... таких как файлы XML

Дальнейшая ссылка для посещения

Есть прямое чистое доступное решение. Используйте метод wholeTextFiles (). Это займет каталог и сформирует пару ключ-значение. Возвращенная СДР будет парой СДР. Найдите ниже описание из документов Spark :

SparkContext.wholeTextFiles позволяет читать каталог, содержащий несколько небольших текстовых файлов, и возвращает каждый из них в виде пар (имя файла, содержимое). Это отличается от textFile, который будет возвращать одну запись на строку в каждом файле

ПОПРОБУЙТЕ ЭТОТ интерфейс, используемый для записи DataFrame во внешние системы хранения (например, файловые системы, хранилища ключей и т. Д.). Используйте DataFrame.write () для доступа к этому.

Новое в версии 1.4.

csv (путь, режим = Нет, сжатие = Нет, септ = Нет, кавычка = Нет, escape = Нет, заголовок = Нет, nullValue = Нет, escapeQuotes = Нет, quoteAll = Нет, dateFormat = Нет, timestampFormat = Нет) Сохраняет содержимое DataFrame в формате CSV по указанному пути.

Параметры: путь - путь в любом режиме файловой системы, поддерживаемом Hadoop, - определяет поведение операции сохранения, когда данные уже существуют.

append: добавить содержимое этого DataFrame к существующим данным. перезаписать: перезаписать существующие данные. игнорировать: игнорировать эту операцию, если данные уже существуют. ошибка (случай по умолчанию): выдает исключение, если данные уже существуют. сжатие - кодек сжатия, используемый при сохранении в файл. Это может быть одно из известных сокращенных имен без учета регистра (none, bzip2, gzip, lz4, snappy и deflate). sep - устанавливает отдельный символ в качестве разделителя для каждого поля и значения. Если None установлен, он использует значение по умолчанию,. цитата - устанавливает единственный символ, используемый для экранирования значений в кавычках, где разделитель может быть частью значения. Если None установлен, он использует значение по умолчанию ". Если вы хотите отключить кавычки, вам нужно установить пустую строку. Escape - устанавливает единственный символ, используемый для экранирования кавычек внутри уже заключенного в кавычки значения. Если None установлен , он использует значение по умолчанию, \ escapeQuotes - флаг, указывающий, должны ли значения, содержащие кавычки, всегда заключаться в кавычки. Если None установлен, он использует значение по умолчанию true, экранируя все значения, содержащие символ кавычки. quoteAll - Флаг, указывающий, должны ли все значения всегда заключаться в кавычки. Если None установлен, он использует значение по умолчанию false, только экранируя значения, содержащие символ кавычки. header - записывает имена столбцов в качестве первой строки. Если None установлен, он использует значение по умолчанию, false. nullValue - устанавливает строковое представление нулевого значения. Если None установлен, он использует значение по умолчанию, пустую строку. dateFormat - устанавливает строку, которая указывает формат даты. Пользовательские форматы даты следуют форматам в java.text.SimpleDateFormat. Это относится к типу даты. Если None установлен, он использует значение по умолчанию, yyyy-MM-dd. timestampFormat - устанавливает строку, которая указывает формат отметки времени. Пользовательские форматы даты следуют форматам в java.text.SimpleDateFormat. Это относится к метке времени. Если значение None установлено, используется значение по умолчанию: yyyy-MM-dd'T'HH: mm: ss.SSSZZ.

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