You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by al...@apache.org on 2018/02/17 01:47:32 UTC
[2/3] kudu git commit: [spark] enable scan locality by default
[spark] enable scan locality by default
Commit 3abca98c5 introduced support to take advantage of scan locality
in Spark integration, however this feature is not enabled by default.
This patch defaults scan locality to use closest replica. And if the
client configures to not to use the default, then only list leader
replica as the perferred location to schedule a task.
Change-Id: Ifc94e9ce52877a88f96a06599a8657da929b6126
Reviewed-on: http://gerrit.cloudera.org:8080/9163
Tested-by: Kudu Jenkins
Reviewed-by: Dan Burkert <da...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/a1854325
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/a1854325
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/a1854325
Branch: refs/heads/master
Commit: a1854325b1c769bafcd20504b5f50c72195bc790
Parents: fc2e878
Author: hahao <ha...@cloudera.com>
Authored: Tue Jan 30 13:54:44 2018 -0800
Committer: Hao Hao <ha...@cloudera.com>
Committed: Fri Feb 16 23:26:33 2018 +0000
----------------------------------------------------------------------
.../org/apache/kudu/spark/kudu/DefaultSource.scala | 4 ++--
.../main/scala/org/apache/kudu/spark/kudu/KuduRDD.scala | 12 ++++++++++--
2 files changed, 12 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/a1854325/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/DefaultSource.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/DefaultSource.scala b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/DefaultSource.scala
index 7987bf8..58beec7 100644
--- a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/DefaultSource.scala
+++ b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/DefaultSource.scala
@@ -71,7 +71,7 @@ class DefaultSource extends RelationProvider with CreatableRelationProvider
val operationType = getOperationType(parameters.getOrElse(OPERATION, "upsert"))
val faultTolerantScanner = Try(parameters.getOrElse(FAULT_TOLERANT_SCANNER, "false").toBoolean)
.getOrElse(false)
- val scanLocality = getScanLocalityType(parameters.getOrElse(SCAN_LOCALITY, "leader_only"))
+ val scanLocality = getScanLocalityType(parameters.getOrElse(SCAN_LOCALITY, "closest_replica"))
new KuduRelation(tableName, kuduMaster, faultTolerantScanner,
scanLocality, operationType, None)(sqlContext)
@@ -107,7 +107,7 @@ class DefaultSource extends RelationProvider with CreatableRelationProvider
val operationType = getOperationType(parameters.getOrElse(OPERATION, "upsert"))
val faultTolerantScanner = Try(parameters.getOrElse(FAULT_TOLERANT_SCANNER, "false").toBoolean)
.getOrElse(false)
- val scanLocality = getScanLocalityType(parameters.getOrElse(SCAN_LOCALITY, "leader_only"))
+ val scanLocality = getScanLocalityType(parameters.getOrElse(SCAN_LOCALITY, "closest_replica"))
new KuduRelation(tableName, kuduMaster, faultTolerantScanner,
scanLocality, operationType, Some(schema))(sqlContext)
http://git-wip-us.apache.org/repos/asf/kudu/blob/a1854325/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduRDD.scala
----------------------------------------------------------------------
diff --git a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduRDD.scala b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduRDD.scala
index 5a4ed8b..ddfdf46 100644
--- a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduRDD.scala
+++ b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduRDD.scala
@@ -62,8 +62,16 @@ class KuduRDD private[kudu] (val kuduContext: KuduContext,
val tokens = builder.build().asScala
tokens.zipWithIndex.map {
case (token, index) =>
- new KuduPartition(index, token.serialize(),
- token.getTablet.getReplicas.asScala.map(_.getRpcHost).toArray)
+ // Only list the leader replica as the preferred location if
+ // replica selection policy is leader only, to take advantage
+ // of scan locality.
+ var locations: Array[String] = null
+ if (scanLocality == ReplicaSelection.LEADER_ONLY) {
+ locations = Array(token.getTablet.getLeaderReplica.getRpcHost)
+ } else {
+ locations = token.getTablet.getReplicas.asScala.map(_.getRpcHost).toArray
+ }
+ new KuduPartition(index, token.serialize(), locations)
}.toArray
}