Как загрузить файлы tar.gz в потоковые наборы данных?

Я не думаю, что чтение файлов tar.gz возможно в Spark (см. Чтение целых текстовых файлов из сжатие в Spark или поддержку gzip в Spark для некоторых идей).

Spark поддерживает файлы gzip, но они не рекомендуются, так как не могут быть разделены и приводят к созданию одного раздела (что, в свою очередь, делает Spark практически бесполезным).

Чтобы загружать сжатые файлы в Spark Structured Streaming, вы должны указать шаблон пути, чтобы файлы включались в загрузку, скажем, zsessionlog*.csv.gz или подобное. В противном случае csv загружает только файлы CSV.

Если вы настаиваете на том, чтобы Spark Structured Streaming обрабатывал файлы tar.gz, вы можете написать собственные данные потоковой передачи Source, чтобы выполнить не-tar.gz.

Учитывая, что файлы gzip не рекомендуются в качестве формата данных в Spark, вся идея использования Spark Structured Streaming не имеет особого смысла.

Я решил часть чтения файлов .tar.gz (.tgz) следующим образом: сжатие-кодек-для-hadoop/" rel="nofollow noreferrer">сайт Я создал свой собственный кодек TGZ

final class DecompressTgzCodec extends CompressionCodec {
  override def getDefaultExtension: String = ".tgz"

  override def createOutputStream(out: OutputStream): CompressionOutputStream = ???
  override def createOutputStream(out: OutputStream, compressor: Compressor): CompressionOutputStream = ???
  override def createCompressor(): Compressor = ???
  override def getCompressorType: Class[_ <: Compressor] = ???

  override def createInputStream(in: InputStream): CompressionInputStream = {
    new TarDecompressorStream(new TarArchiveInputStream(new GzipCompressorInputStream(in)))
  }
  override def createInputStream(in: InputStream, decompressor: Decompressor): CompressionInputStream = createInputStream(in)

  override def createDecompressor(): Decompressor = null
  override def getDecompressorType: Class[_ <: Decompressor] = null

  final class TarDecompressorStream(in: TarArchiveInputStream) extends DecompressorStream(in) {
    def updateStream(): Unit = {
      // still have data in stream -> done
      if (in.available() <= 0) {
        // create stream content from following tar elements one by one
        in.getNextTarEntry()
      }
    }

    override def read: Int = {
      checkStream()
      updateStream()
      in.read()
    }

    override def read(b: Array[Byte], off: Int, len: Int): Int = {
      checkStream()
      updateStream()
      in.read(b, off, len)
    }

    override def resetState(): Unit = {}
  }
}

И зарегистрировал его для использования искрой.

val conf = new SparkConf()
conf.set("spark.hadoop.io.compression.codecs", classOf[DecompressTgzCodec].getName)

val spark = SparkSession
  .builder()
  .master("local[*]")
  .config(conf)
  .appName("Streaming Example")
  .getOrCreate()

Работает именно так, как я хотел.