在独立/主从火花壳中从拼花地板读取时的不同行为

Question

这是我用来从 Scala 中的 Parquet 读取数据帧的较大代码片段。

case class COOMatrix(row: Seq[Long], col: Seq[Long], data: Seq[Double])
def buildMatrix(cooMatrixFields: DataFrame) = {
  val cooMatrices = cooMatrixFields map {
    case Row(r,c,d) => COOMatrix(r.asInstanceOf[Seq[Long]], c.asInstanceOf[Seq[Long]], d.asInstanceOf[Seq[Double]])
  }
  val matEntries = cooMatrices.zipWithIndex.flatMap {
    case (cooMat, matIndex) =>
      val rowOffset = cooMat.row.distinct.size
      val colOffset = cooMat.col.distinct.size
      val cooMatRowShifted = cooMat.row.map(rowEntry => rowEntry + rowOffset * matIndex)
      val cooMatColShifted = cooMat.col.map(colEntry => colEntry + colOffset * matIndex)
      (cooMatRowShifted, cooMatColShifted, cooMat.data).zipped.map {
        case (i, j, value) => MatrixEntry(i, j, value)
      }
  }
  new CoordinateMatrix(matEntries)
}

val C_entries = sqlContext.read.load(s"${dataBaseDir}/C.parquet")
val C = buildMatrix(C_entries)

我的代码在本地 Spark 上下文中运行时成功执行。

在独立集群上，相同的代码在到达强制其实际从 Parquet 读取的操作时立即失败。已正确检索数据帧的架构：

C_entries: org.apache.spark.sql.DataFrame = [C_row: array<bigint>, C_col: array<bigint>, C_data: array<double>]

但是执行程序在执行此行时崩溃 val C = buildMatrix(C_entries) ，但有以下例外：

java.lang.ExceptionInInitializerError
    at $line39.$read$$iwC.<init>(<console>:7)
    at $line39.$read.<init>(<console>:61)
    at $line39.$read$.<init>(<console>:65)
    at $line39.$read$.<clinit>(<console>)
    at $line67.$read$$iwC.<init>(<console>:7)
    at $line67.$read.<init>(<console>:24)
    at $line67.$read$.<init>(<console>:28)
    at $line67.$read$.<clinit>(<console>)
    at $line68.$read$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$3.apply(<console>:63)
    at $line68.$read$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$3.apply(<console>:62)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
    at org.apache.spark.util.Utils$.getIteratorSize(Utils.scala:1597)
    at org.apache.spark.rdd.ZippedWithIndexRDD$$anonfun$2.apply(ZippedWithIndexRDD.scala:52)
    at org.apache.spark.rdd.ZippedWithIndexRDD$$anonfun$2.apply(ZippedWithIndexRDD.scala:52)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.NullPointerException
    at $line4.$read$$iwC$$iwC.<init>(<console>:15)
    at $line4.$read$$iwC.<init>(<console>:24)
    at $line4.$read.<init>(<console>:26)
    at $line4.$read$.<init>(<console>:30)
    at $line4.$read$.<clinit>(<console>)
    ... 22 more

不确定它是否相关，但在增加日志详细程度的同时，我注意到了以下异常：

16/03/07 20:59:38 INFO GenerateUnsafeProjection: Code generated in 157.285464 ms
16/03/07 20:59:38 DEBUG ExecutorClassLoader: Did not load class org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection from REPL class server at http://155.198.193.158:32862
java.lang.ClassNotFoundException: Class file not found at URL http://155.198.193.158:32862/org/apache/spark/sql/catalyst/expressions/GeneratedClass%24SpecificUnsafeProjection.class

我已经为独立集群尝试了不同的配置：

• 主站，1 个从机和火花壳在我的笔记本电脑上运行
• 主站和 1 个从站分别运行在不同的机器上，我的笔记本电脑上的火花壳
• 一台机器上的主站和火花壳，另一台机器上的一台从站。

我从默认属性开始，并演变成一个更复杂的属性文件，但没有取得更大的成功：

spark.driver.memory                4g
spark.rpc=netty
spark.eventLog.enabled             true
spark.eventLog.dir                 file:///mnt/fastmp/spark_workdir/logs
spark.driver.extraJavaOptions      -Xmx20480m -XX:MaxPermSize=2048m -XX:ReservedCodeCacheSize=2048m
spark.shuffle.service.enabled      true
spark.shuffle.consolidateFiles     true
spark.sql.parquet.binaryAsString   true
spark.speculation                  false
spark.rpc.timeout                 1000
spark.rdd.compress true
spark.core.connection.ack.wait.timeout 600
spark.driver.maxResultSize         0
spark.task.maxFailures             3
spark.shuffle.io.maxRetries        3

我正在运行 spark-1.6.0-bin-hadoop2.6 的预构建版本。此部署不涉及 HDFS，所有 Parquet 文件都存储在所有计算机可用的共享挂载 （CephFS） 上。

我怀疑这与底层文件系统有关，因为我代码的另一部分在本地和独立模式下都可以很好地读取不同的 Parquet 文件。

Answer 1

TL;DR：将代码打包为jar

为了记录的目的，这个问题似乎与使用独立集群有关。

完全相同的代码适用于以下设置：

• 火花壳和主站在同一台机器上
• 在 YARN（AWS EMR 集群）上运行并从 S3 读取镶木地板文件

随着对独立设置日志的更多挖掘，问题似乎与类服务器的此异常有关：

INFO GenerateUnsafeProjection: Code generated in 157.285464 ms
DEBUG ExecutorClassLoader: Did not load class org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection from REPL class server at http://155.198.193.158:32862
java.lang.ClassNotFoundException: Class file not found at URL http://155.198.193.158:32862/org/apache/spark/sql/catalyst/expressions/GeneratedClass%24SpecificUnsafeProjection.class

我的理解是，spark-shell启动一个HTTP服务器（码头），以便将它从REPL中的代码生成的类提供给worker。

就我而言，许多课程都成功送达（我什至设法通过 telnet 检索了一些课程）。但是，类服务器找不到类GeneratedClass（及其所有内部类）。

日志中出现的典型错误消息是：

DEBUG Server: RESPONSE /org/apache/spark/sql/catalyst/expressions/GeneratedClass.class  404 handled=true

我的想法是，它与 master 和 spark-shell 在同一服务器上工作，因为它们在同一 JVM 中运行，因此即使 HTTP 传输失败，也可以找到该类。

到目前为止，我发现的唯一成功的解决方案是构建一个jar包并使用spark-shell的--jars选项或将其作为参数传递给spark-submit。