Apache Geode with Spark

在一些特定场景，例如streamingRDD需要和历史数据进行join从而获得一些profile信息，此时形成较小的新数据RDD和很大的历史RDD的join。
Spark中直接join实际上效率不高：

• RDD没有索引，join操作实际上是相互join的RDD进行hash然后shuffle到一起；

实际上，如果历史数据的RDD有索引，我们可以循环遍历streaming中的每一条数据，并向历史数据发送point query，即loop + indexed get。Streaming的数据是小数据，这样坐的性能会高很多。（这种小数据和大量历史数据的join模式在物联网/互联网场景下很常见）

另外，

• spark中的RDD是只读的，增量信息无法直接更新到历史RDD中

虽然我们可以使用streaming的窗口操作来缓存一定量的历史数据，但这会增加业务逻辑的复杂度。

IndexedRDD能够解决上述的两个问题，即对RDD内存数据建立索引，并且可以更新RDD。但是IndexRDD不支持事务，如果需要对同一个key做更新就存在数据更新冲突，导致数据不一致。另外，IndexRDD单纯是RDD的数据结构和接口的增强，不支持Spark之外的组件对其的访问。

本文将介绍基于Apache Geode和Spark相结合：

• 基于Geode的RDD借助Geode的内存数据存储和数据索引，其join操作是loop + indexed get方式，可以提高流数据和历史数据相join的效率；
• Geode 是目前性能和生产可用性最高的IMDG之一，基本满足ACID；
• Spark 中通过GeodeRDD的写操作实际上是将数据写入Geode，我们还可以通过JDBC等方式访问数据，甚至进行OLAP操作。

Geode和spark版本选择

• geode
  https://github.com/apache/geode/blob/rel/v1.1.1/
  这个版本包含了spark-connector。
• spark
  选择1.6，如果选择2.0则需要手动编译基于scala2.10的包。

Geode-Spark connector编译

需要手动编译spark-connector，参照GitHub上的流程操作即可。
https://github.com/apache/geode/blob/rel/v1.1.1/geode-spark-connector/doc/1_building.md
最终会编译三个文件：
The following jar files will be created:

• geode-spark-connector/target/scala-2.10/geode-spark-connector_2.10-0.5.0.jar
• geode-functions/target/scala-2.10/geode-functions_2.10-0.5.0.jar
• geode-spark-demos/target/scala-2.10/geode-spark-demos_2.10-0.5.0.jar

启动geode并创建region

Start Geode cluster with 1 locator and 2 servers:

gfsh
gfsh>start locator --name=locator1 --port=55221
gfsh>start server --name=server1 --locators=localhost[55221] --server-port=0
gfsh>start server --name=server2 --locators=localhost[55221] --server-port=0

Then create two demo regions:

gfsh>create region --name=str_str_region --type=PARTITION --key-constraint=java.lang.String --value-constraint=java.lang.String
gfsh>create region --name=int_str_region --type=PARTITION --key-constraint=java.lang.Integer --value-constraint=java.lang.String

Deploy Spark Geode Connector's geode-function jar (geode-functions_2.10-0.5.0.jar):

gfsh>deploy --jar=<path to connector project>/geode-functions/target/scala-2.10/geode-functions_2.10-0.5.0.jar

Spark 启动

官网下载spark1.6.0-bin-hadoop2.6。解压后运行./sbin/start-all。

进入spark-shell并引入Geode包

export GEDE=<path to geode>/apache-geode-1.1.1/

spark-shell --master spark://Dings-MacBook-Pro.local:7077 --jars /Users/dingbingbing/hon/geode/geode/geode-spark-connector/geode-spark-connector/target/scala-2.10/geode-spark-connector_2.10-0.5.0.jar,/Users/dingbingbing/hon/geode/geode/geode-spark-connector/geode-functions/target/scala-2.10/geode-functions_2.10-0.5.0.jar,$GEDE/lib/activation-1.1.jar,$GEDE/lib/antlr-2.7.7.jar,$GEDE/lib/commons-beanutils-1.8.3.jar,$GEDE/lib/commons-io-2.4.jar,$GEDE/lib/commons-lang-2.5.jar,$GEDE/lib/commons-logging-1.2.jar,$GEDE/lib/commons-modeler-2.0.jar,$GEDE/lib/fastutil-7.0.2.jar,$GEDE/lib/findbugs-annotations-1.3.9-1.jar,$GEDE/lib/geode-common-1.1.1.jar,$GEDE/lib/geode-core-1.1.1.jar,$GEDE/lib/geode-cq-1.1.1.jar,$GEDE/lib/geode-dependencies.jar,$GEDE/lib/geode-json-1.1.1.jar,$GEDE/lib/geode-lucene-1.1.1.jar,$GEDE/lib/geode-old-client-support-1.1.1.jar,$GEDE/lib/geode-rebalancer-1.1.1.jar,$GEDE/lib/geode-wan-1.1.1.jar,$GEDE/lib/geode-web-1.1.1.jar,$GEDE/lib/gfsh-dependencies.jar,$GEDE/lib/jackson-annotations-2.8.0.jar,$GEDE/lib/jackson-core-2.8.2.jar,$GEDE/lib/jackson-databind-2.8.2.jar,$GEDE/lib/jansi-1.8.jar,$GEDE/lib/javax.mail-api-1.4.5.jar,$GEDE/lib/javax.resource-api-1.7.jar,$GEDE/lib/javax.servlet-api-3.1.0.jar,$GEDE/lib/javax.transaction-api-1.2.jar,$GEDE/lib/jetty-http-9.3.6.v20151106.jar,$GEDE/lib/jetty-io-9.3.6.v20151106.jar,$GEDE/lib/jetty-security-9.3.6.v20151106.jar,$GEDE/lib/jetty-server-9.3.6.v20151106.jar,$GEDE/lib/jetty-servlet-9.3.6.v20151106.jar,$GEDE/lib/jetty-util-9.3.6.v20151106.jar,$GEDE/lib/jetty-webapp-9.3.6.v20151106.jar,$GEDE/lib/jetty-xml-9.3.6.v20151106.jar,$GEDE/lib/jgroups-3.6.10.Final.jar,$GEDE/lib/jline-2.12.jar,$GEDE/lib/jna-4.0.0.jar,$GEDE/lib/jopt-simple-5.0.1.jar,$GEDE/lib/log4j-api-2.6.1.jar,$GEDE/lib/log4j-core-2.6.1.jar,$GEDE/lib/log4j-jcl-2.6.1.jar,$GEDE/lib/log4j-jul-2.6.1.jar,$GEDE/lib/log4j-slf4j-impl-2.6.1.jar,$GEDE/lib/lucene-analyzers-common-6.0.0.jar,$GEDE/lib/lucene-core-6.0.0.jar,$GEDE/lib/lucene-queries-6.0.0.jar,$GEDE/lib/lucene-queryparser-6.0.0.jar,$GEDE/lib/mx4j-3.0.1.jar,$GEDE/lib/mx4j-remote-3.0.1.jar,$GEDE/lib/mx4j-tools-3.0.1.jar,$GEDE/lib/netty-all-4.0.4.Final.jar,$GEDE/lib/ra.jar,$GEDE/lib/shiro-core-1.3.1.jar,$GEDE/lib/slf4j-api-1.7.21.jar,$GEDE/lib/snappy-0.4.jar,$GEDE/lib/spring-aop-4.3.2.RELEASE.jar,$GEDE/lib/spring-beans-4.3.2.RELEASE.jar,$GEDE/lib/spring-context-4.3.2.RELEASE.jar,$GEDE/lib/spring-core-4.3.2.RELEASE.jar,$GEDE/lib/spring-expression-4.3.2.RELEASE.jar,$GEDE/lib/spring-shell-1.2.0.RELEASE.jar,$GEDE/lib/spring-web-4.3.2.RELEASE.jar --conf spark.geode.locators=localhost[55221]

Check Geode locator property in the Spark shell:

scala> sc.getConf.get("spark.geode.locators")
res0: String = localhost[55221]

测试代码及原理简介

Geode可以认为是类似hdfs/hbase的数据集，不同的是：

• 基于Geode数据形成的RDD可以被修改；
• 普通的RDD可以和Geode Region数据快速join；

使用Geode Spark Connector的代码中首先import一下org.apache.geode.spark.connector._。引入所有的implicit函数。

scala> import org.apache.geode.spark.connector._
import org.apache.geode.spark.connector._

Save Pair RDD to Geode

In the Spark shell, create a simple pair RDD and save it to Geode:

scala> val data = Array(("1", "one"), ("2", "two"), ("3", "three"))
data: Array[(String, String)] = Array((1,one), (2,two), (3,three))

scala> val distData = sc.parallelize(data)
distData: org.apache.spark.rdd.RDD[(String, String)] = ParallelCollectionRDD[0] at parallelize at <console>:14

scala> distData.saveToGemfire("str_str_region")
15/02/17 07:11:54 INFO DAGScheduler: Job 0 finished: runJob at GemFireRDDFunctions.scala:29, took 0.341288 s

此时Geode中相应region就有了刚才save的数据了gfsh:

gfsh>query --query="select key,value from /str_str_region.entries"

Result     : true
startCount : 0
endCount   : 20
Rows       : 3

key | value
--- | -----
1   | one
3   | three
2   | two

NEXT_STEP_NAME : END

Save Non-Pair RDD to Geode

Saving non-pair RDD to Geode requires an extra function that converts each
element of RDD to a key-value pair. Here's sample session in Spark shell:

scala> val data2 = Array("a","ab","abc")
data2: Array[String] = Array(a, ab, abc)

scala> val distData2 = sc.parallelize(data2)
distData2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at <console>:17

scala> distData2.saveToGemfire("int_str_region", e => (e.length, e))
[info 2015/02/17 12:43:21.174 PST <main> tid=0x1]
...
15/02/17 12:43:21 INFO DAGScheduler: Job 0 finished: runJob at GemFireRDDFunctions.scala:52, took 0.251194 s

Verify the result with gfsh:

gfsh>query --query="select key,value from /int_str_region.entrySet"

Result     : true
startCount : 0
endCount   : 20
Rows       : 3

key | value
--- | -----
2   | ab
3   | abc
1   | a

NEXT_STEP_NAME : END

Expose Geode Region As RDD

The same API is used to expose both replicated and partitioned region as RDDs.

scala> val rdd = sc.geodeRegion[String, String]("str_str_region")
rdd: org.apache.geode.spark.connector.rdd.GemFireRDD[String,String] = GemFireRDD[2] at RDD at GemFireRDD.scala:19

scala> rdd.foreach(println)
(1,one)
(3,three)
(2,two)

scala> val rdd2 = sc.geodeRegion[Int, String]("int_str_region")
rdd2: org.apache.geode.spark.connector.rdd.GemFireRDD[Int,String] = GemFireRDD[3] at RDD at GemFireRDD.scala:19

scala> rdd2.foreach(println)
(2,ab)
(1,a)
(3,abc)

Join性能测试（极简单案例）

// 10万条数据
val device_id = sc.parallelize((1 to 100000).map(i => ("device_"+i, "device_id = "+ i + ", value="+(new scala.util.Random().nextInt()))))
// save to Geode
device_id.saveToGeode("str_str_region")
// 1000条数据作为新增数据
val new_rdd = sc.parallelize((4000 to 5000).map(i => ("device_"+i, "device_id = "+ i + ", value="+(new scala.util.Random().nextInt()))))
// 新数据和Geode中十万条join
new_rdd.joinGeodeRegion("str_str_region", p => p._1).count()
// 新增数据和十万条数据的RDD join
new_rdd.join(device_id).count()

10万条数据的性能差别有将近10倍。

具体来说，RDD跟Geode Regioin的join是循环+get操作，类似于map-only 的join。具体代码参见GeodeJoinRdd.scala

private def computeWithoutFunc(split: Partition, context: TaskContext, region: Region[K, V]): Iterator[(T, V)] = {
    val leftPairs = left.iterator(split, context).toList.asInstanceOf[List[(K, _)]]
    val leftKeys = leftPairs.map { case (k, v) => k}.toSet
    // Note: get all will return (key, null) for non-exist entry, so remove those entries
    val rightPairs = region.getAll(leftKeys).filter { case (k, v) => v != null}
    leftPairs.filter{case (k, v) => rightPairs.contains(k)}
             .map {case (k, v) => ((k, v).asInstanceOf[T], rightPairs.get(k).get)}.toIterator
  }

而RDD跟RDD的普通join操作需要数据的shuffle，会带来很多额外的开销。如下图所示。

可以推断一下，在一些特定场景，例如streamingRDD需要和历史数据进行join从而获得一些profile信息，此时形成较小的新数据RDD和很大的历史RDD的join。此时loop + index get的性能会高很多。这种小数据和大量历史数据的join模式在物联网/互联网场景下很常见。

此外IndexedRdd也可以作为一个备选方案。但是IndexedRdd无法向Geode这样能够被Spark世界之外访问，只能作为提高spark计算的一种方案.