IndexedRDDAMPLabAnkur Dave提出,它是ImmutabilityFine-Grained updates的精妙结合。IndexedRDD是一个基于RDDKey-Value Store,扩展自RDD[(K, V)],可以在IndexRDD上进行高效的查找、更新以及删除。由于其并没有合并到 Spark 的主项目分支,所以在使用时需要引入特别的对其的支持。

IndexedRDD的详细分析

这里主要是记录引进 IndexedRDD 之后项目出现的各种错误及解决过程,目前关于 IndexedRDD 的文章不多,百度出来的与搭环境有关系的也就十几篇左右,出现错误更是无解,所以特此记录一下填坑之路。

开始引入 IndexedRDD

参见 Github 的说明,在 build.sbt 中添加:

1
2
3
4
5
6
//这句很关键
resolvers += "Spark Packages Repo" at "http://dl.bintray.com/spark-packages/maven"
libraryDependencies += "amplab" % "spark-indexedrdd" % "0.3"
//顺带引入GraphX
libraryDependencies += "org.apache.spark" %% "spark-graphx" % "2.2.0"

编译错误

注意:这里出现了一个天坑,总是编译(包含IndexedRDD时)出错的问题

历经解决过程:

解决措施一

明确 scalaspark 版本的对照关系,版本确定为:
scala-2.11.8
spark-core-2.1.0(graphx同2.1.0)

上述版本是 spark-rdd 代码库中 build.sbt 的版本,详见 Github-spark-indexedrdd

明确 spark-indexedrdd 版本
注意,maven源 的版本只有 0.1 0.2 0.3 0.4.0 这四个,Github代码库中的实例程序推荐的是 0.3

但是编译时会出现如下错误:

1
2
3
4
5
6
Run:
18/05/22 01:29:47 WARN ClosureCleaner: Expected a closure; got edu.berkeley.cs.amplab.spark.indexedrdd.IndexedRDD$MultiputZipper
Exception in thread "main" java.lang.NoSuchMethodError: org.apache.spark.SparkContext.runJob
Sbt shell:
could not find implicit value for evidence parameter of type edu.berkeley.cs.amplab.spark.indexedrdd.KeySerializer[Long]

解决措施二

这时看到了 源库 的这个 Issue
于是将 spark-indexedrdd 改为 0.4.0 版本,注意是三位数字

然后继续编译仍然出同样的错(没效果)

解决措施三

这时,又看见了 这个问题 其错误跟咱们的不一样,但是格式太像了,然后看他的解决方案:

报错分析:这种异常的发生通常是因为程序需要一个隐式参数 (implicit parameter)
方法的定义中有个 [R: TypeInformation] ,但程序并没有指定任何有关隐式参数的定义,编译代码无法创建 TypeInformation ,所以出现上面提到的异常信息。
解决方案:
1) 我们可以直接在代码里面加上以下的代码:
  implicit val typeInfo = TypeInformation.of(classOf[Int])
然后再去编译代码就不会出现上面的异常。
2) 但是这并不是Flink推荐我们去做的,推荐的做法是在代码中引入一下包:
  import org.apache.flink.streaming.api.scala._
如果数据是有限的(静态数据集),我们可以引入以下包:
  import org.apache.flink.api.scala._
然后即可解决上面的异常信息。

同样的思路,翻过头来看,自己项目里的 import 确实少了一个!

1
2
3
4
import edu.berkeley.cs.amplab.spark.indexedrdd.IndexedRDD
// 下面这个不引入也不会报错,但是会编译出错
// 还要注意顺序,上下颠倒IDEA会自动省略
import edu.berkeley.cs.amplab.spark.indexedrdd.IndexedRDD._

同时还要注意,之前改为 0.4.0 版本是对的,如果换做 0.3 ,此时还是会编译出错

结论

综上,IndexedRDD 环境(示例运行正常)应该如下:

scala-2.11.8 
spark-core-2.1.0
graphx-2.1.0(非必须)
spark-indexedrdd-0.4.0

build.sbt 文件:

1
2
3
4
5
6
7
name := "VISNWK"
version := "0.1"
scalaVersion := "2.11.8"
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"
libraryDependencies += "org.apache.spark" %% "spark-graphx" % "2.1.0"
resolvers += "Spark Packages Repo" at "http://dl.bintray.com/spark-packages/maven"
libraryDependencies += "amplab" % "spark-indexedrdd" % "0.4.0"

IndexedRDD demo(IDEA环境下):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
import edu.berkeley.cs.amplab.spark.indexedrdd.IndexedRDD //缺一不可
import edu.berkeley.cs.amplab.spark.indexedrdd.IndexedRDD._ //缺一不可
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import scala.util.Random
object graphxDemo {
def main(args: Array[String]) {
//设置运行环境
val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
val sc = new SparkContext(conf)
// Create an RDD of key-value pairs with Long keys.
val rdd = sc.parallelize((1 to 1000000).map(x => (x.toLong, 0)))
// Construct an IndexedRDD from the pairs, hash-partitioning and indexing
// the entries.
val indexed = IndexedRDD(rdd).cache()
// Perform a point update.
val indexed2 = indexed.put(1234L, 10873).cache()
// Perform a point lookup. Note that the original IndexedRDD remains
// unmodified.
indexed2.get(1234L) // => Some(10873)
indexed.get(1234L) // => Some(0)
// Efficiently join derived IndexedRDD with original.
val indexed3 = indexed.innerJoin(indexed2) { (id, a, b) => b }.filter(_._2 != 0)
indexed3.collect // => Array((1234L, 10873))
// Perform insertions and deletions.
val indexed4 = indexed2.put(-100L, 111).delete(Array(998L, 999L)).cache()
indexed2.get(-100L) // => None
indexed4.get(-100L) // => Some(111)
indexed2.get(999L) // => Some(0)
indexed4.get(999L) // => None
sc.stop()
}
}

其他错误

注意,之前还出现过 Apache Spark: Java.Lang.NoSuchMethodError .RddToPairRDDFunctions 这个错误,但是今天明确版本后就没有复现,所以该错误八成是因为版本不兼容的缘故,总之还是版本不兼容引起的编译错误。

还有这个错误 unresolved dependency: com.ankurdave#part_2.10;0.1,之前是使用 Sbt 和 Maven 混用,然后用 Maven 添加的 spark-indexedrdd 才出现的这个错误,在改用 Sbt 单一管理依赖后该错误也没有复现。