将一个大的 tar.gz 文件分成多个较小的 tar.gz 文件

Question

在 spark 中处理大于 1gb 的 tar.gz 文件时，我得到 OutOfMemoryError。

为了克服这个错误，我尝试使用“split”命令将 tar.gz 拆分为多个部分，结果发现每个拆分都不是 tar.gz 本身，因此不能这样处理。

dir=/dbfs/mnt/data/temp
b=524288000
for file in /dbfs/mnt/data/*.tar.gz; 
do 
a=$(stat -c%s "$file");
if [[ "$a" -gt "$b" ]] ; then 
split -b 500M -d --additional-suffix=.tar.gz $file "${file%%.*}_part"
mv $file $dir
fi
done

尝试处理拆分文件时出错

Caused by: java.io.EOFException
    at org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream.read(GzipCompressorInputStream.java:281)
    at java.io.BufferedInputStream.read1(BufferedInputStream.java:284)
    at java.io.BufferedInputStream.read(BufferedInputStream.java:345)
    at org.apache.commons.compress.archivers.tar.TarArchiveInputStream.read(TarArchiveInputStream.java:590)
    at org.apache.commons.io.input.ProxyInputStream.read(ProxyInputStream.java:98)
    at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:284)
    at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:326)
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:178)
    at java.io.InputStreamReader.read(InputStreamReader.java:184)
    at java.io.Reader.read(Reader.java:140)
    at org.apache.commons.io.IOUtils.copyLarge(IOUtils.java:2001)
    at org.apache.commons.io.IOUtils.copyLarge(IOUtils.java:1980)
    at org.apache.commons.io.IOUtils.copy(IOUtils.java:1957)
    at org.apache.commons.io.IOUtils.copy(IOUtils.java:1907)
    at org.apache.commons.io.IOUtils.toString(IOUtils.java:778)
    at org.apache.commons.io.IOUtils.toString(IOUtils.java:803)
    at linea3796c25fa964697ba042965141ff28825.$read$$iw$$iw$$iw$$iw$$iw$$iw$Unpacker$$anonfun$apply$1.apply(command-2152765781429277:33)
    at linea3796c25fa964697ba042965141ff28825.$read$$iw$$iw$$iw$$iw$$iw$$iw$Unpacker$$anonfun$apply$1.apply(command-2152765781429277:31)
    at scala.collection.immutable.Stream$$anonfun$map$1.apply(Stream.scala:418)
    at scala.collection.immutable.Stream$$anonfun$map$1.apply(Stream.scala:418)
    at scala.collection.immutable.Stream$Cons.tail(Stream.scala:1233)
    at scala.collection.immutable.Stream$Cons.tail(Stream.scala:1223)
    at scala.collection.immutable.Stream.foreach(Stream.scala:595)
    at scala.collection.TraversableOnce$class.toMap(TraversableOnce.scala:316)
    at scala.collection.AbstractTraversable.toMap(Traversable.scala:104)
    at linea3796c25fa964697ba042965141ff28825.$read$$iw$$iw$$iw$$iw$$iw$$iw$Unpacker$.apply(command-2152765781429277:34)
    at linea3796c25fa964697ba042965141ff28827.$read$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(command-2152765781429278:3)
    at linea3796c25fa964697ba042965141ff28827.$read$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(command-2152765781429278:3)

我的 tar.gz 文件大小高达 4gb，每个文件最多可包含 7000 个 json 文档，大小从 1mb 到 50mb 不等。

如果我想将大的 tar.gz 文件分成更小的 tar.gz 文件，我唯一的选择是解压缩，然后以某种方式根据文件大小或文件数重新压缩？ - “是这样吗？”

Answer 1

普通 gzip 文件不可拆分。 GZip Tar 档案更难处理。 Spark 可以处理 gzip 压缩的 json 文件，但不能处理 gzip 压缩的 tar 文件和 tar 文件。 Spark 可以处理每个最大约 2GB 的二进制文件。 Spark 可以处理连接在一起的 JSON

我建议使用 Pandas UDF 或 .pipe() 运算符来处理每个 tar gzip 压缩文件（每个工作人员一个）。每个工作人员将以流式方式解压缩、解压缩和处理每个 JSON 文档，从不填满内存。希望您有足够的源文件来并行运行它并看到速度加快。

您可能想要探索流式传输方法，以将压缩的 JSON 文件增量交付到 ADLS Gen 2 / S3，并使用 Databricks 自动加载器功能在文件到​​达后立即加载和处理。

这个问题的答案How to load tar.gz files in streaming datasets? 似乎很有希望。