Bagaimana cara membaca banyak file teks menjadi satu RDD?

Saya ingin membaca banyak file teks dari lokasi HDFS dan melakukan pemetaan pada iterasi menggunakan percikan.

JavaRDD<String> records = ctx.textFile(args[1], 1); mampu membaca hanya satu file pada satu waktu.

Saya ingin membaca lebih dari satu file dan memprosesnya sebagai RDD tunggal. Bagaimana?

Anda dapat menentukan seluruh direktori, menggunakan wildcard dan bahkan CSV dari direktori dan wildcard. Misalnya:

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

Seperti yang ditunjukkan oleh Nick Chammas, ini adalah paparan dari Hadoop FileInputFormatdan oleh karena itu ini juga bekerja dengan Hadoop (dan Scalding).

Gunakan unionsebagai berikut:

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

Maka itu bigRddadalah RDD dengan semua file.

Anda dapat menggunakan satu panggilan textFile untuk membaca banyak file. Scala:

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

Anda bisa menggunakan ini

Pertama, Anda bisa mendapatkan Buffer / Daftar S3 Paths:

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
  }

Sekarang Lewati objek Daftar ini ke potongan kode berikut, catatan: sc adalah objek SQLContext

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

Sekarang Anda mendapatkan RDD Unified akhir yaitu df

Opsional, Dan Anda juga dapat mempartisi ulang dalam BigRDD tunggal

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

Partisi ulang selalu berfungsi: D

Di PySpark, saya telah menemukan cara tambahan yang berguna untuk mem-parsing file. Mungkin ada padanan dalam Scala, tapi saya tidak cukup nyaman dengan terjemahan yang berfungsi. Ini adalah, pada dasarnya, panggilan textFile dengan penambahan label (dalam contoh di bawah ini kunci = nama file, nilai = 1 baris dari file).

TextFile "berlabel"

memasukkan:

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 dengan setiap entri yang berisi tuple menggunakan nama file-sebagai-kunci dan dengan nilai = setiap baris file. (Secara teknis, menggunakan metode ini Anda juga dapat menggunakan kunci yang berbeda di samping nama file sebenarnya - mungkin representasi hashing untuk menghemat memori). yaitu.

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

Anda juga dapat menggabungkan kembali baik sebagai daftar baris:

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

Atau gabungkan kembali seluruh file kembali ke string tunggal (dalam contoh ini hasilnya sama dengan apa yang Anda dapatkan dari wholeTextFiles, tetapi dengan string "file:" yang dilepas dari filepathing.):

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

kamu bisa memakai

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

di sini Anda akan mendapatkan jalur file Anda dan konten file itu. sehingga Anda dapat melakukan tindakan apa pun dari seluruh file pada waktu yang menghemat overhead

Semua jawaban benar sc.textFile

Saya hanya ingin tahu mengapa tidak wholeTextFilesMisalnya, dalam hal ini ...

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

satu batasannya adalah, kita harus memuat file kecil jika tidak kinerja akan buruk dan dapat menyebabkan OOM.

Catatan :

• Seluruh file harus sesuai dengan memori
• Baik untuk format file yang TIDAK dapat dibagi berdasarkan baris ... seperti file XML

Referensi lebih lanjut untuk dikunjungi

Ada solusi bersih lurus ke depan yang tersedia. Gunakan metode wholeTextFiles (). Ini akan mengambil direktori dan membentuk pasangan nilai kunci. RDD yang dikembalikan akan menjadi pasangan RDD. Temukan deskripsi di bawah ini dari Spark docs :

SparkContext.wholeTextFiles memungkinkan Anda membaca direktori yang berisi banyak file teks kecil, dan mengembalikan masing-masing sebagai pasangan (nama file, konten). Ini berbeda dengan textFile, yang akan mengembalikan satu catatan per baris di setiap file

COBALAH Antarmuka INI digunakan untuk menulis DataFrame ke sistem penyimpanan eksternal (mis. Sistem file, penyimpanan nilai kunci, dll). Gunakan DataFrame.write () untuk mengakses ini.

Baru dalam versi 1.4.

csv (jalur, mode = Tidak Ada, kompresi = Tidak ada, sep = Tidak ada, kutipan = Tidak ada, melarikan diri = Tidak ada, header = Tidak ada, nullValue = Tidak ada, escapeQuotes = Tidak ada, quoteAll = Tidak ada, dateFormat = Tidak ada, timestampFormat = Tidak Ada) Menyimpan konten DataFrame dalam format CSV di jalur yang ditentukan.

Parameter: jalur - jalur dalam mode sistem file apa pun yang didukung Hadoop - menentukan perilaku operasi penyimpanan saat data sudah ada.

append: Tambahkan konten DataFrame ini ke data yang ada. menimpa: Menimpa data yang ada. abaikan: Abaikan operasi ini secara diam-diam jika data sudah ada. error (case default): Melempar pengecualian jika data sudah ada. kompresi - kompresi codec untuk digunakan saat menyimpan ke file. Ini bisa menjadi salah satu dari nama pendek yang tidak peka terhadap huruf yang diketahui (tidak ada, bzip2, gzip, lz4, tajam dan mengempis). sep - menetapkan karakter tunggal sebagai pemisah untuk setiap bidang dan nilai. Jika Tidak ada yang diset, ini menggunakan nilai default,,. kutipan - mengatur karakter tunggal yang digunakan untuk keluar dari nilai yang dikutip di mana pemisah dapat menjadi bagian dari nilai. Jika Tidak ada yang ditetapkan, ini menggunakan nilai default, ". Jika Anda ingin mematikan kutipan, Anda perlu mengatur string kosong. Escape - mengatur karakter tunggal yang digunakan untuk keluar dari kutipan di dalam nilai yang sudah dikutip. Jika Tidak ada yang ditetapkan , menggunakan nilai default, \ escapeQuotes - Bendera yang menunjukkan apakah nilai yang mengandung tanda kutip harus selalu dilampirkan dalam tanda kutip. Jika Tidak Ada yang disetel, ini menggunakan nilai default true, keluar dari semua nilai yang mengandung karakter kutipan. quoteAll - Bendera yang menunjukkan apakah semua nilai harus selalu dilampirkan dalam tanda kutip. Jika Tidak ada yang ditetapkan, ini menggunakan nilai default false, hanya melarikan diri nilai yang mengandung karakter kutipan. header - menulis nama kolom sebagai baris pertama. Jika Tidak ada yang diset, ini menggunakan nilai default, false. nullValue - mengatur representasi string dari nilai null. Jika Tidak ada yang diset, ini menggunakan nilai default, string kosong. dateFormat - mengatur string yang menunjukkan format tanggal. Format tanggal khusus mengikuti format di java.text.SimpleDateFormat. Ini berlaku untuk tipe tanggal. Jika Tidak Ada yang diset, ini menggunakan nilai nilai default, yyyy-MM-dd. timestampFormat - mengatur string yang menunjukkan format timestamp. Format tanggal khusus mengikuti format di java.text.SimpleDateFormat. Ini berlaku untuk jenis cap waktu. Jika Tidak Ada yang diset, ini menggunakan nilai nilai default, yyyy-MM-dd'T'HH: mm: ss.SSSZZ.

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