eclipse - sc.TextFile ("") 在 Eclipse 中工作但不在 JAR 中

Question

我正在编写将在 hadoop 集群中的代码，但首先，我使用本地文件在本地对其进行测试。该代码在 Eclipse 中运行良好，但是当我使用 SBT(使用 spark lib 等)制作一个巨大的 JAR 时，该程序一直运行到 textFile(path) 我的代码是:

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.log4j.{Level, Logger}
import org.joda.time.format.DateTimeFormat
import org.apache.spark.rdd.RDD
import scala.collection.mutable.ArrayBuffer




object TestCRA2 {

    val conf = new SparkConf()
      .setMaster("local")
      .setAppName("Test")
      .set("spark.driver.memory", "4g")
      .set("spark.executor.memory", "4g")
    val context = new SparkContext(conf)//.master("local")
    val rootLogger = Logger.getRootLogger()
    rootLogger.setLevel(Level.ERROR)

    def TimeParse1(path: String) : RDD[(Int,Long,Long)] = {
        val data = context.textFile(path).map(_.split(";"))
        return data
    }

    def main(args: Array[String]) {

        val data = TimeParse1("file:///home/quentin/Downloads/CRA") 
    }
}

这是我的错误:

Exception in thread "main" java.io.IOException: No FileSystem for scheme: file
    at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:2586)
    at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2593)
    at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:91)
    at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2632)
    at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2614)
    at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:370)
    at org.apache.hadoop.fs.FileSystem.getLocal(FileSystem.java:341)
    at org.apache.spark.SparkContext$$anonfun$hadoopFile$1.apply(SparkContext.scala:1034)
    at org.apache.spark.SparkContext$$anonfun$hadoopFile$1.apply(SparkContext.scala:1029)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.SparkContext.withScope(SparkContext.scala:701)
    at org.apache.spark.SparkContext.hadoopFile(SparkContext.scala:1029)
    at org.apache.spark.SparkContext$$anonfun$textFile$1.apply(SparkContext.scala:832)
    at org.apache.spark.SparkContext$$anonfun$textFile$1.apply(SparkContext.scala:830)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.SparkContext.withScope(SparkContext.scala:701)
    at org.apache.spark.SparkContext.textFile(SparkContext.scala:830)
    at main.scala.TestCRA2$.TimeParse1(TestCRA.scala:37)
    at main.scala.TestCRA2$.main(TestCRA.scala:84)
    at main.scala.TestCRA2.main(TestCRA.scala)

我无法将我的文件放入 JAR 中，因为它们在集群 hadoop 中并且正在 Eclipse 上运行。

这是我的 build.sbt :

name := "BloomFilters"

version := "1.0"

scalaVersion := "2.11.6"

libraryDependencies += "org.apache.spark" %% "spark-core" % "2.2.0"


libraryDependencies += "joda-time" % "joda-time" % "2.9.3"

assemblyMergeStrategy in assembly := {
 case PathList("META-INF", xs @ _*) => MergeStrategy.discard
 case x => MergeStrategy.first
}

如果我不像这样执行我的assemblyMergeStrategy，我就会遇到一堆合并错误。

实际上我需要像这样更改我的 build.sbt :

name := "BloomFilters"

version := "1.0"

scalaVersion := "2.11.6"

libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.2.0"

libraryDependencies += "joda-time" % "joda-time" % "2.9.3"

assemblyMergeStrategy in assembly := {
    case PathList("META-INF", xs @ _*) =>
      (xs map {_.toLowerCase}) match {

        case "services" :: xs => MergeStrategy.first

        case _ => MergeStrategy.discard
 }
 case x => MergeStrategy.first
}

谢谢@lyomi

Answer 1

您的 sbt 程序集 可能忽略了一些必需的文件。具体来说，Hadoop 的 FileSystem 类依赖于一种服务发现机制，该机制在类路径中查找所有 META-INFO/services/org.apache.hadoop.fs.FileSystem 文件。

在 Eclipse 上没问题，因为每个 JAR 都有相应的文件，但在 uber-jar 中可能覆盖了其他文件，导致 file: 方案无法识别。

在您的 SBT 设置中，添加以下内容以连接服务发现文件而不是丢弃其中的一些文件。

val defaultMergeStrategy: String => MergeStrategy = { 
  case PathList("META-INF", xs @ _*) =>
    (xs map {_.toLowerCase}) match {
      // ... possibly other settings ...
      case "services" :: xs =>
        MergeStrategy.filterDistinctLines
      case _ => MergeStrategy.deduplicate
  }
  case _ => MergeStrategy.deduplicate
}

参见 README.md of sbt-assembly了解更多信息。