Skip to content

Added Region caching #306

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Kamil-Krynicki wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Added Region caching #306

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
78594f5
Added meta cache
Feb 25, 2019
20c81bf
Added configuration to meta cache
Feb 25, 2019
51658ce
Reverted to byte-compatibility
Feb 28, 2019
26e5ea5
Reverted to byte-compatibility
Mar 1, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions core/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,11 @@
<artifactId>avro</artifactId>
<version>${avro.version}</version>
</dependency>
<dependency>
<groupId>com.github.blemale</groupId>
<artifactId>scaffeine_2.11</artifactId>
<version>2.5.0</version>
</dependency>
<dependency>
<groupId>org.scalatest</groupId>
<artifactId>scalatest_${scala.binary.version}</artifactId>
Expand Down
20 changes: 18 additions & 2 deletions core/src/main/scala/org/apache/spark/sql/execution/datasources/hbase/HBaseTableScan.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,8 @@ import org.apache.spark.sql.execution.datasources.hbase.types.{SHCDataType, SHCD
import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types._
import org.apache.spark.util.ShutdownHookManager

import com.github.blemale.scaffeine.{LoadingCache, Scaffeine}
import scala.concurrent.duration._
import scala.collection.mutable
import scala.util.matching.Regex

Expand All @@ -51,6 +52,21 @@ private[hbase] case class HBaseScanPartition(
scanRanges: Array[ScanRange[Array[Byte]]],
tf: SerializedTypedFilter) extends Partition

object MetaCache {
private def sparkConf = SparkEnv.get.conf

private[this] val useCache = sparkConf.getBoolean(SparkHBaseConf.MetaCacheEnabled, SparkHBaseConf.defaultMetaCacheEnabled)
private[this] val expirySeconds = sparkConf.getTimeAsSeconds(SparkHBaseConf.MetaCacheDuration, SparkHBaseConf.defaultMetaCacheDuration)

private[this] val cache: LoadingCache[HBaseRelation, RegionResource] = Scaffeine()
.expireAfterWrite(Duration(expirySeconds, SECONDS))
.build((relation: HBaseRelation) => RegionResource(relation))

private[this] val regionHook : HBaseRelation => RegionResource = if (useCache) r => this.cache.get(r) else r => RegionResource(r)

def get(relation: HBaseRelation) : RegionResource = this.regionHook(relation)
}

private[hbase] class HBaseTableScanRDD(
relation: HBaseRelation,
requiredColumns: Array[String],
Expand All @@ -63,7 +79,7 @@ private[hbase] class HBaseTableScanRDD(
override def getPartitions: Array[Partition] = {
val hbaseFilter = HBaseFilter.buildFilters(filters, relation)
var idx = 0
val r = RegionResource(relation)
val r = MetaCache.get(relation)
logDebug(s"There are ${r.size} regions")
val ps = r.flatMap { x=>
// HBase take maximum as empty byte array, change it here.
Expand Down
7 changes: 7 additions & 0 deletions core/src/main/scala/org/apache/spark/sql/execution/datasources/hbase/SparkHBaseConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ package org.apache.spark.sql.execution.datasources.hbase

import org.apache.hadoop.conf.Configuration
import org.apache.spark.sql.execution.datasources.hbase.types._
import scala.concurrent.duration._


object SparkHBaseConf {
Expand All @@ -38,6 +39,12 @@ object SparkHBaseConf {
var defaultBulkGetSize = 100
var CachingSize = "spark.hbase.connector.cacheSize"
var defaultCachingSize = 100

var MetaCacheEnabled = "spark.hbase.connector.meta.cache.enabled"
var defaultMetaCacheEnabled = false
var MetaCacheDuration = "spark.hbase.connector.meta.cache.expiry"
var defaultMetaCacheDuration = "5min"

// in milliseconds
val connectionCloseDelay = 10 * 60 * 1000

Expand Down