apache-spark - Spark - 忽略损坏的文件

Question

在我们管理的 ETL 过程中，我们有时会收到损坏的文件。我们尝试了这个 Spark 配置，它似乎有效(Spark 作业没有失败，因为损坏的文件被丢弃):

spark.sqlContext.setConf("spark.sql.files.ignoreCorruptFiles", "true")

但我不知道是否有办法知道哪些文件被忽略了。有没有办法获取这些文件名？

提前致谢

Answer 1

一种方法是查看您的执行程序日志。如果您在 spark 配置中将以下配置设置为 true。

RDD: spark.files.ignoreCorruptFiles数据框:spark.sql.files.ignoreCorruptFiles

然后 spark 会将损坏的文件作为 WARN 消息记录在您的执行程序日志中。

这是来自 Spark 的代码片段:

      if (ignoreCorruptFiles) {
        currentIterator = new NextIterator[Object] {
          // The readFunction may read some bytes before consuming the iterator, e.g.,
          // vectorized Parquet reader. Here we use lazy val to delay the creation of
          // iterator so that we will throw exception in `getNext`.
          private lazy val internalIter = readCurrentFile()

          override def getNext(): AnyRef = {
            try {
              if (internalIter.hasNext) {
                internalIter.next()
              } else {
                finished = true
                null
              }
            } catch {
              // Throw FileNotFoundException even `ignoreCorruptFiles` is true
              case e: FileNotFoundException => throw e
              case e @ (_: RuntimeException | _: IOException) =>
                logWarning(
                  s"Skipped the rest of the content in the corrupted file: $currentFile", e)
                finished = true
                null
            }
          }