找不到数据源：parquet，使用 sbt 程序集构建时

Question

我有以下代码：

val testRDD: RDD[(String, Vector)] = sc.parallelize(testArray)

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext.implicits._

val df = testRDD.toDF()

df.write.parquet(path)

使用以下 build.sbt：

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.1"
libraryDependencies += "org.apache.spark" %% "spark-sql" % "1.6.1"
libraryDependencies += "org.apache.spark" %% "spark-mllib" % "1.6.1"

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

当我使用 sbt-assembly 构建它时（我有 addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.12.0")），然后我运行它，我得到一个错误：

Exception in thread "main" java.lang.ClassNotFoundException: Failed to find data source: parquet. Please find packages at http://spark-packages.org
    at org.apache.spark.sql.execution.datasources.ResolvedDataSource$.lookupDataSource(ResolvedDataSource.scala:77)
    at org.apache.spark.sql.execution.datasources.ResolvedDataSource$.apply(ResolvedDataSource.scala:219)
    at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:148)
    at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:139)
    at org.apache.spark.sql.DataFrameWriter.parquet(DataFrameWriter.scala:334)
    at InductionService.Application$.main(ParquetTest.scala:65)

但是，如果我使用 IntelliJ Idea 构建它（正常构建，而不是像 sbt 程序集那样的胖 JAR 类型的东西），并在该 IDE 中调试它，它实际上可以工作。很明显，我使用 sbt 程序集构建它的方式有问题，但我不知道如何修复它。有什么想法吗？

我怀疑 build.sbt 中的 merge-inf 丢弃代码可能是原因，但我需要该代码，否则我无法使用 sbt-assembly 构建。 （它抱怨重复......）

Answer 1

我遇到了问题。 META-INF 中的 services 文件夹存在一些合并问题。 我可以通过在 MergeStrategy 上添加规则来解决这个问题：

case n if n.contains("services") => MergeStrategy.concat

这就是我所拥有的，现在它可以工作了：

val meta = """META.INF(.)*""".r
assemblyMergeStrategy in assembly := {
  case PathList("javax", "servlet", xs @ _*) => MergeStrategy.first
  case PathList(ps @ _*) if ps.last endsWith ".html" => MergeStrategy.first
  case n if n.contains("services") => MergeStrategy.concat
  case n if n.startsWith("reference.conf") => MergeStrategy.concat
  case n if n.endsWith(".conf") => MergeStrategy.concat
  case meta(_) => MergeStrategy.discard
  case x => MergeStrategy.first
}

Answer 2

上述解决方案并不理想，它部分正确，您确实需要合并 META-INF/service/.... 带丢弃 META-INF 中的其他带，但上述解决方案使用 case n if n.contains("service") 完成它，这是一个过于笼统的条件。 如果任何重复文件，不仅在 META-INF 中包含非常常见的“服务”一词，那么该文件将被连接起来，包括类文件。 如果您的程序或其依赖项之一包含类似 com/amazonaws/services/s3/model/AmazonS3Exception 的类，就像我的一样，它会将它们连接起来，导致：

java.lang.ClassFormatError: Extra bytes at the end of class file com/amazonaws/services/s3/model/AmazonS3Exception

最好尽可能限制 concat 子句，给出一个子句“concat META-INF/services，丢弃任何其他 META-INF”，这是一种方法：

assemblyMergeStrategy in assembly := {
  case PathList("META-INF", "services",  _*)  => MergeStrategy.concat
  case PathList("META-INF", _*) => MergeStrategy.discard
  case _ => MergeStrategy.first

}