Spark/Hadoop 对大型 LZO 文件抛出异常

Question

我正在对存储在 S3 中的一些 LZO 压缩日志文件运行 EMR Spark 作业。有多个日志文件存储在同一个文件夹中，例如：

...
s3://mylogfiles/2014-08-11-00111.lzo
s3://mylogfiles/2014-08-11-00112.lzo
...

在 spark-shell 中，我正在运行一项计算文件中行数的作业。如果我为每个文件单独计算行数，则没有问题，例如像这样：

// Works fine
...
sc.textFile("s3://mylogfiles/2014-08-11-00111.lzo").count()
sc.textFile("s3://mylogfiles/2014-08-11-00112.lzo").count()
...

如果我使用通配符以单行方式加载所有文件，则会出现两种异常。

// One-liner throws exceptions
sc.textFile("s3://mylogfiles/*.lzo").count()

例外情况是：

java.lang.InternalError: lzo1x_decompress_safe returned: -6
    at com.hadoop.compression.lzo.LzoDecompressor.decompressBytesDirect(Native Method)

和

java.io.IOException: Compressed length 1362309683 exceeds max block size 67108864 (probably corrupt file)
    at com.hadoop.compression.lzo.LzopInputStream.getCompressedData(LzopInputStream.java:291)

在我看来，解决方案是由最后一个例外给出的文本暗示的，但我不知道如何继续。 LZO 文件的大小是否有限制，或者有什么问题？

我的问题是：我能否运行 Spark 查询，将所有 LZO 压缩文件加载到 S3 文件夹中，而不会出现与 I/O 相关的异常？

有 66 个文件，每个文件大约 200MB。

编辑： 仅在使用 Hadoop2 核心库（ami 3.1.0）运行 Spark 时才会发生异常。使用 Hadoop1 核心库（ami 2.4.5）运行时，一切正常。两种情况均使用 Spark 1.0.1 进行了测试。

Answer 1

kgeyti 的回答很好，但是：

LzoTextInputFormat 会影响性能，因为它会检查每个 LZO 文件的 .index 文件。这对于 S3 上的许多 LZO 文件来说尤其痛苦（我经历了长达几分钟的延迟，这是由对 S3 的数千个请求引起的）。

如果您事先知道您的 LZO 文件是不可可拆分的，那么更高效的解决方案是创建自定义的不可拆分输入格式：

import org.apache.hadoop.fs.Path
import org.apache.hadoop.mapreduce.JobContext
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat

class NonSplittableTextInputFormat extends TextInputFormat {
    override def isSplitable(context: JobContext, file: Path): Boolean = false
}

并像这样读取文件：

context.newAPIHadoopFile("s3://mylogfiles/*.lzo",
  classOf[NonSplittableTextInputFormat],
  classOf[org.apache.hadoop.io.LongWritable],
  classOf[org.apache.hadoop.io.Text])
.map(_._2.toString)

Answer 2

我自己还没有遇到这个具体问题，但看起来.textFile 期望文件是可拆分的，就像 Cedrik 的 Hive 坚持使用 CombineFileInputFormat 的问题

您可以index your lzo files，或尝试使用LzoTextInputFormat - 我很想知道这在 EMR 上是否更有效：

sc.newAPIHadoopFile("s3://mylogfiles/*.lz", 
    classOf[com.hadoop.mapreduce.LzoTextInputFormat],
    classOf[org.apache.hadoop.io.LongWritable],
    classOf[org.apache.hadoop.io.Text])
  .map(_._2.toString) // if you just want a RDD[String] without writing a new InputFormat
  .count

Answer 3

昨天我们在 EMR 集群上部署了 Hive，并且在 S3 中的一些 LZO 文件遇到了同样的问题，这些文件已被另一个非 EMR 集群毫无问题地采取。在对日志进行了一些挖掘之后，我注意到地图任务以 250MB 块读取 S3 文件，尽管这些文件绝对不可拆分。

原来参数ma​​preduce.input.fileinputformat.split.maxsize设置为250000000~250MB。这导致 LZO 从文件中打开流，最终导致 LZO 块损坏。

我将参数 ma​​preduce.input.fileinputformat.split.maxsize=2000000000 设置为我们输入数据的最大文件大小，现在一切正常。

我不确定这与 Spark 有什么关系，但更改 InputFormat 可能会有所帮助，这似乎首先是问题所在，正如 How Amazon EMR Hive Differs from Apache Hive 中提到的那样。