Spark - чтение сжатых файлов без расширения файла

Question

У меня есть ведро S3, которое заполнено файлами Gz, которые не имеют расширения файла. Например, s3://mybucket/1234502827-34231

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

Я думаю, что решение может быть sc.binaryFiles и разархивировать файл вручную.

Другая возможность - выяснить, как sc.textFile находит формат файла. Я не понимаю, как работают эти classOf[].

def textFile(
     path: String, 
     minPartitions: Int = defaultMinPartitions): RDD[String] = withScope { 
    assertNotStopped() 
    hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text], 
     minPartitions).map(pair => pair._2.toString).setName(path) 
    } 

Answer 1

Можете ли вы попытаться объединить нижеследующее решение для файлов ZIP с библиотекой gzipFileInputFormat?

здесь - How to open/stream .zip files through Spark? Вы можете видеть, как сделать это с помощью ZIP:

rdd1 = sc.newAPIHadoopFile("/Users/myname/data/compressed/target_file.ZIP", ZipFileInputFormat.class, Text.class, Text.class, new Job().getConfiguration()); 

gzipFileInputFormat:

https://github.com/bsankaran/internet_routing/blob/master/hadoop-tr/src/main/java/edu/usc/csci551/tools/GZipFileInputFormat.java

Некоторые подробности о newAPIHadoopFile() можно найти здесь: http://spark.apache.org/docs/latest/api/python/pyspark.html

Answer 2

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

import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream 
import org.apache.spark.input.PortableDataStream 
import scala.util.Try 
import java.nio.charset._ 

def extractBSM(ps: PortableDataStream, n: Int = 1024) = Try { 
    val gz = new GzipCompressorInputStream(ps.open) 
    Stream.continually { 
    // Read n bytes 
    val buffer = Array.fill[Byte](n)(-1) 
    val i = gz.read(buffer, 0, n) 
    (i, buffer.take(i)) 
    } 
    // Take as long as we've read something 
    .takeWhile(_._1 > 0) 
    .map(_._2) 
    .flatten 
    .toArray 
} 
def decode(charset: Charset = StandardCharsets.UTF_8)(bytes: Array[Byte]) = new String(bytes, StandardCharsets.UTF_8) 
val inputFile = "s3://my-bucket/157c96bd-fb21-4cc7-b340-0bd4b8e2b614" 
val rdd = sc.binaryFiles(inputFile).flatMapValues(x => extractBSM(x).toOption).map(x => decode()(x._2)) 
val rdd2 = rdd.flatMap { x => x.split("\n") } 
rdd2.take(10).foreach(println)