在 Spark 上运行 Tika 的类路径问题

Question

我尝试在 Tika 中处理一堆文件。文件的数量有数千个，所以我决定构建一个文件的 RDD 并让 Spark 分配工作负载。不幸的是，我收到了多个 NoClassDefFound 异常。

这是我的 sbt 文件：

name := "TikaFileParser"
version := "0.1"
scalaVersion := "2.11.7"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.5.1" % "provided"
libraryDependencies += "org.apache.tika" % "tika-core" % "1.11"
libraryDependencies += "org.apache.tika" % "tika-parsers" % "1.11"
libraryDependencies += "org.apache.hadoop" % "hadoop-client" % "2.7.1" % "provided"

这是我的程序集.sbt

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.1")

这是源文件：

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.input.PortableDataStream
import org.apache.tika.metadata._
import org.apache.tika.parser._
import org.apache.tika.sax.WriteOutContentHandler
import java.io._

object TikaFileParser {

  def tikaFunc (a: (String, PortableDataStream)) = {

    val file : File = new File(a._1.drop(5))
    val myparser : AutoDetectParser = new AutoDetectParser()
    val stream : InputStream = new FileInputStream(file)
    val handler : WriteOutContentHandler = new WriteOutContentHandler(-1)
    val metadata : Metadata = new Metadata()
    val context : ParseContext = new ParseContext()

    myparser.parse(stream, handler, metadata, context)

    stream.close

    println(handler.toString())
    println("------------------------------------------------")
  }


  def main(args: Array[String]) {

    val filesPath = "/home/user/documents/*"
    val conf = new SparkConf().setAppName("TikaFileParser")
    val sc = new SparkContext(conf)
    val fileData = sc.binaryFiles(filesPath)
    fileData.foreach( x => tikaFunc(x))
  }
}

我正在运行这个

spark-submit --driver-memory 2g --class TikaFileParser --master local[4]
             /path/to/TikaFileParser-assembly-0.1.jar

并得到java.lang.NoClassDefFoundError: org/apache/cxf/jaxrs/ext/multipart/ContentDisposition，它是解析器的依赖项。出于好奇，我将包含此类的 jar 添加到 Spark 的 --jars 选项并再次运行。这次我得到了一个新的NoClassDefFoundError（不记得是哪个了，还有一个 Tika 依赖）。

我已经在这里 (Apache Tika 1.11 on Spark NoClassDeftFoundError) 发现了一个类似的问题，解决方案是构建一个胖罐。但是我想知道是否有其他方法可以解决依赖问题？

顺便说一句：我在没有 Spark 的情况下尝试了这个 sn-p（所以只需使用带有文件名和 foreach 循环的数组并相应地更改 tikaFunc 签名。我在没有任何参数的情况下运行它并且它运行良好。

编辑：现在更新了 sn-ps 以用于 sbt 程序集。

Answer 1

我已经在这里发现了一个类似的问题（Apache Tika 1.11 on Spark NoClassDeftFoundError），解决方案是构建一个胖罐。但是我想知道是否有其他方法可以解决依赖问题？

找到所有依赖项并将它们添加到--jars。您可以使用https://github.com/jrudolph/sbt-dependency-graph 来完成。但我不明白为什么你更喜欢这个而不是构建一个将它们组合在一起的罐子。

我在没有任何参数的情况下运行它，它运行良好。

SBT 已经确保您拥有类路径上的所有依赖项，但 Spark 不使用 SBT 来运行您的程序。

Answer 2

我想纠正@flowit 的答案，因为它让我陷入了漫长的一天调查。

答案的问题在于合并策略，它丢弃了每个META-INF 目录。然而，这也将摆脱 Tika 注册 i.a 的 META-INF/services 目录。它的解析器。

使用合并策略，您可以在已接受的答案或其他飞来飞去的 Stackoverflow 答案中找到该策略，您最终会得到空内容，因为 Tika 将默认为 EmptyParser。因此，如果您尝试解析任何内容，Tika 将无法解析解析器。见https://tika.apache.org/1.21/configuring.html#Static。

对我来说，解决方案是（我猜是使用更新的 sbt 语法）：

assemblyMergeStrategy in assembly := {
  case PathList("META-INF", xs @ _*) =>
    (xs map {_.toLowerCase}) match {
      case "services" :: xs => MergeStrategy.concat // Tika uses the META-INF/services to register its parsers statically, don't discard it
      case _ => MergeStrategy.discard
    }
  case x => MergeStrategy.first
}

Answer 3

问题来自 jar 中的版本不匹配。我决定使用以下 sbt 文件来解决我的问题：

name := "TikaFileParser"
version := "0.1"
scalaVersion := "2.11.7"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.5.1" % "provided"
libraryDependencies += "org.apache.tika" % "tika-core" % "1.11"
libraryDependencies += "org.apache.tika" % "tika-parsers" % "1.11"
libraryDependencies += "org.apache.hadoop" % "hadoop-client" % "2.7.1" % "provided"

mergeStrategy in assembly <<= (mergeStrategy in assembly) { (old) =>
  {
    case PathList("META-INF", xs @ _*) => MergeStrategy.discard
    case _     => MergeStrategy.first
  }
}