devbox
羊村懒王
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Apache Geode with Spark

作者：羊村懒王 | 2024-02-27 22:18:34

赞

踩

spark geode

在一些特定场景，例如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-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:

  1. gfsh
  2. gfsh>start locator --name=locator1 --port=55221
  3. gfsh>start server --name=server1 --locators=localhost[55221] --server-port=0
  4. gfsh>start server --name=server2 --locators=localhost[55221] --server-port=0

Then create two demo regions:

  1. gfsh>create region --name=str_str_region --type=PARTITION --key-constraint=java.lang.String --value-constraint=java.lang.String
  2. 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包

  1. export GEDE=<path to geode>/apache-geode-1.1.1/
  2.  
  3. 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:

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

测试代码及原理简介

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

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

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

  1. scala> import org.apache.geode.spark.connector._
  2. 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:

  1. scala> val data = Array(("1", "one"), ("2", "two"), ("3", "three"))
  2. data: Array[(String, String)] = Array((1,one), (2,two), (3,three))
  3.  
  4. scala> val distData = sc.parallelize(data)
  5. distData: org.apache.spark.rdd.RDD[(String, String)] = ParallelCollectionRDD[0] at parallelize at <console>:14
  6.  
  7. scala> distData.saveToGemfire("str_str_region")
  8. 15/02/17 07:11:54 INFO DAGScheduler: Job 0 finished: runJob at GemFireRDDFunctions.scala:29, took 0.341288 s

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

  1. gfsh>query --query="select key,value from /str_str_region.entries"
  2.  
  3. Result     : true
  4. startCount : 0
  5. endCount   : 20
  6. Rows       : 3
  7.  
  8. key | value
  9. --- | -----
  10. 1   | one
  11. 3   | three
  12. 2   | two
  13.  
  14. 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:

  1. scala> val data2 = Array("a","ab","abc")
  2. data2: Array[String] = Array(a, ab, abc)
  3.  
  4. scala> val distData2 = sc.parallelize(data2)
  5. distData2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at <console>:17
  6.  
  7. scala> distData2.saveToGemfire("int_str_region", e => (e.length, e))
  8. [info 2015/02/17 12:43:21.174 PST <main> tid=0x1]
  9. ...
  10. 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:

  1. gfsh>query --query="select key,value from /int_str_region.entrySet"
  2.  
  3. Result     : true
  4. startCount : 0
  5. endCount   : 20
  6. Rows       : 3
  7.  
  8. key | value
  9. --- | -----
  10. 2   | ab
  11. 3   | abc
  12. 1   | a
  13.  
  14. NEXT_STEP_NAME : END

Expose Geode Region As RDD

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

  1. scala> val rdd = sc.geodeRegion[String, String]("str_str_region")
  2. rdd: org.apache.geode.spark.connector.rdd.GemFireRDD[String,String] = GemFireRDD[2] at RDD at GemFireRDD.scala:19
  3.  
  4. scala> rdd.foreach(println)
  5. (1,one)
  6. (3,three)
  7. (2,two)
  8.  
  9. scala> val rdd2 = sc.geodeRegion[Int, String]("int_str_region")
  10. rdd2: org.apache.geode.spark.connector.rdd.GemFireRDD[Int,String] = GemFireRDD[3] at RDD at GemFireRDD.scala:19
  11.  
  12. scala> rdd2.foreach(println)
  13. (2,ab)
  14. (1,a)
  15. (3,abc)

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

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

两种join操作的时间对比

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

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

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

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

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

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

转载于:https://www.cnblogs.com/luweiseu/p/7698986.html

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/羊村懒王/article/detail/156230
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号