[HADOOP] Spark에서 압축 된 전체 텍스트 파일 읽기

Spark에서 압축 된 전체 텍스트 파일 읽기

다음과 같은 문제점이 있습니다. HDFS에 저장된 여러 파일을 포함하는 압축 된 디렉토리가있는 디렉토리가 있다고 가정합니다. 나는 유형 T의 몇몇 객체들로 구성된 RDD를 만들고 싶다.

context = new JavaSparkContext(conf);
JavaPairRDD<String, String> filesRDD = context.wholeTextFiles(inputDataPath);

JavaPairRDD<String, String> filesRDD = context.wholeTextFiles(inputDataPath);
JavaRDD<T> processingFiles = filesRDD.map(fileNameContent -> {
    // The name of the file
    String fileName = fileNameContent._1();
    // The content of the file
    String content = fileNameContent._2();

    // Class T has a constructor of taking the filename and the content of each
    // processed file (as two strings)
    T t = new T(content, fileName);

    return t;
});

이제 inputDataPath가 파일을 포함하고있는 디렉토리 일 때, 이것은 완벽하게 잘 작동합니다. 예를 들면 다음과 같습니다 :

String inputDataPath =  "hdfs://some_path/*/*/"; // because it contains subfolders

하지만, 여러 파일을 포함하는 tgz가있을 때, 파일 내용 (fileNameContent._2 ())은 쓸모없는 이진 문자열을 가져옵니다 (꽤 기대됩니다). 그래서 비슷한 질문을 찾았지만 똑같은 경우는 아닙니다. 솔루션은 각 압축이 하나의 파일로만 이루어져 있고 제 경우에는 전체 파일로 개별적으로 읽으려는 다른 많은 파일이 있기 때문입니다. wholeTextFiles에 대한 질문을 찾았지만 내 경우에는 작동하지 않습니다.

어떤 아이디어가 이것을 어떻게?

편집하다:

나는 여기서 reader (여기서 testTarballWithFolders ()와 같이 독자를 테스트하려고 시도)를 시도했지만, 호출 할 때마다

TarballReader tarballReader = new TarballReader(fileName);

및 NullPointerException 얻을 :

java.lang.NullPointerException
    at java.util.zip.InflaterInputStream.<init>(InflaterInputStream.java:83)
    at java.util.zip.GZIPInputStream.<init>(GZIPInputStream.java:77)
    at java.util.zip.GZIPInputStream.<init>(GZIPInputStream.java:91)
    at utils.TarballReader.<init>(TarballReader.java:61)
    at main.SparkMain.lambda$0(SparkMain.java:105)
    at main.SparkMain$$Lambda$18/1667100242.call(Unknown Source)
    at org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1.apply(JavaPairRDD.scala:1015)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
    at scala.collection.Iterator$class.foreach(Iterator.scala:727)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
    at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
    at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
    at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
    at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
    at scala.collection.AbstractIterator.to(Iterator.scala:1157)
    at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
    at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
    at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
    at scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
    at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:927)
    at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:927)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

MainSpark의 105 행은 내가 편집 한 게시물의 상단 부분이고 TarballReader의 61 행은

GZIPInputStream gzip = new GZIPInputStream(in);

위의 행에서 입력 스트림에 대한 널 값을 제공합니다.

InputStream in = this.getClass().getResourceAsStream(tarball);

나는 올바른 길을 가고 있는가? 그렇다면 계속하려면 어떻게해야합니까? 이 null 값을 얻는 이유는 무엇이며 어떻게 수정할 수 있습니까?

해결법

1. ==============================

   1.한 가지 가능한 솔루션은 binaryFiles로 데이터를 읽고 내용을 수동으로 추출하는 것입니다.

   한 가지 가능한 솔루션은 binaryFiles로 데이터를 읽고 내용을 수동으로 추출하는 것입니다.

   규모 :

   import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream
   import org.apache.commons.compress.archivers.tar.TarArchiveInputStream
   import org.apache.spark.input.PortableDataStream
   import scala.util.Try
   import java.nio.charset._
   
   def extractFiles(ps: PortableDataStream, n: Int = 1024) = Try {
     val tar = new TarArchiveInputStream(new GzipCompressorInputStream(ps.open))
     Stream.continually(Option(tar.getNextTarEntry))
       // Read until next exntry is null
       .takeWhile(_.isDefined)
       // flatten
       .flatMap(x => x)
       // Drop directories
       .filter(!_.isDirectory)
       .map(e => {
         Stream.continually {
           // Read n bytes
           val buffer = Array.fill[Byte](n)(-1)
           val i = tar.read(buffer, 0, n)
           (i, buffer.take(i))}
         // Take as long as we've read something
         .takeWhile(_._1 > 0)
         .map(_._2)
         .flatten
         .toArray})
       .toArray
   }
   
   def decode(charset: Charset = StandardCharsets.UTF_8)(bytes: Array[Byte]) = 
     new String(bytes, StandardCharsets.UTF_8)
   
   sc.binaryFiles("somePath").flatMapValues(x => 
     extractFiles(x).toOption).mapValues(_.map(decode()))
   

   libraryDependencies += "org.apache.commons" % "commons-compress" % "1.11"
   

   Java의 전체 사용 예제 : https://bitbucket.org/zero323/spark-multifile-targz-extract/src

   파이썬 :

   import tarfile
   from io import BytesIO
   
   def extractFiles(bytes):
       tar = tarfile.open(fileobj=BytesIO(bytes), mode="r:gz")
       return [tar.extractfile(x).read() for x in tar if x.isfile()]
   
   (sc.binaryFiles("somePath")
       .mapValues(extractFiles)
       .mapValues(lambda xs: [x.decode("utf-8") for x in xs]))
   

2. ==============================

   2.허용 된 대답에 대한 약간의 개선은

   허용 된 대답에 대한 약간의 개선은

   옵션 (tar.getNextTarEntry)

   에

   try (tar.getNextTarEntry) .toOption.filter (_! = null)

   견고한 방식으로 조작 된 / 절단 된 .tar.gz와 경쟁해야합니다.

   BTW, 버퍼 배열의 크기에 특별한 것이 있습니까? 평균 파일 크기에 가까울수록 평균 속도는 더 빠를 것입니까, 아니면 제 경우에는 500k일까요? 아니면 더 많은 Java-ish였던 while 루프에 비해 Stream의 오버 헤드가 더 많이 느려지는 것을 느낄 수 있습니다.

from https://stackoverflow.com/questions/36604145/read-whole-text-files-from-a-compression-in-spark by cc-by-sa and MIT license