You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ra...@apache.org on 2019/07/15 12:01:03 UTC
[carbondata] branch master updated: [CARBONDATA-3459] Fixed id
based distribution for showcache command
This is an automated email from the ASF dual-hosted git repository.
ravipesala pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/carbondata.git
The following commit(s) were added to refs/heads/master by this push:
new a682f98 [CARBONDATA-3459] Fixed id based distribution for showcache command
a682f98 is described below
commit a682f98e885bedd3a3d980223937095861c27607
Author: kunal642 <ku...@gmail.com>
AuthorDate: Wed Jul 3 00:53:43 2019 +0530
[CARBONDATA-3459] Fixed id based distribution for showcache command
Problem: Currently tasks are not being fired based on the executor ID because getPrefferedLocation was not overridden.
Solution: override getPreferredLocations in the ShowCache and InvalidateCacheRDD to fire tasks at the appropriate location
This closes #3315
---
.../carbondata/indexserver/DistributedRDDUtils.scala | 6 +++---
.../carbondata/indexserver/DistributedShowCacheRDD.scala | 8 ++++++++
.../indexserver/InvalidateSegmentCacheRDD.scala | 15 ++++++++++++++-
3 files changed, 25 insertions(+), 4 deletions(-)
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedRDDUtils.scala b/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedRDDUtils.scala
index a568153..933ec15 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedRDDUtils.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedRDDUtils.scala
@@ -316,10 +316,10 @@ object DistributedRDDUtils {
if (existingSegmentMapping == null) {
val newSegmentMapping = new ConcurrentHashMap[String, String]()
newSegmentMapping.put(segment.getSegmentNo, s"${newHost}_$newExecutor")
- tableToExecutorMapping.put(tableUniqueName, newSegmentMapping)
+ tableToExecutorMapping.putIfAbsent(tableUniqueName, newSegmentMapping)
} else {
- existingSegmentMapping.put(segment.getSegmentNo, s"${newHost}_$newExecutor")
- tableToExecutorMapping.put(tableUniqueName, existingSegmentMapping)
+ existingSegmentMapping.putIfAbsent(segment.getSegmentNo, s"${newHost}_$newExecutor")
+ tableToExecutorMapping.putIfAbsent(tableUniqueName, existingSegmentMapping)
}
s"executor_${newHost}_$newExecutor"
}
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedShowCacheRDD.scala b/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedShowCacheRDD.scala
index 78b7e72..f1707c6 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedShowCacheRDD.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DistributedShowCacheRDD.scala
@@ -37,6 +37,14 @@ class DistributedShowCacheRDD(@transient private val ss: SparkSession, tableName
}
}.toArray
+ override protected def getPreferredLocations(split: Partition): Seq[String] = {
+ if (split.asInstanceOf[DataMapRDDPartition].getLocations != null) {
+ split.asInstanceOf[DataMapRDDPartition].getLocations.toSeq
+ } else {
+ Seq()
+ }
+ }
+
override protected def internalGetPartitions: Array[Partition] = {
executorsList.zipWithIndex.map {
case (executor, idx) =>
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/InvalidateSegmentCacheRDD.scala b/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/InvalidateSegmentCacheRDD.scala
index c2bd589..750f9d9 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/InvalidateSegmentCacheRDD.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/InvalidateSegmentCacheRDD.scala
@@ -30,7 +30,12 @@ import org.apache.carbondata.spark.rdd.CarbonRDD
class InvalidateSegmentCacheRDD(@transient private val ss: SparkSession, carbonTable: CarbonTable,
invalidSegmentIds: List[String]) extends CarbonRDD[String](ss, Nil) {
- val executorsList: Array[String] = DistributionUtil.getNodeList(ss.sparkContext)
+ val executorsList: Array[String] = DistributionUtil.getExecutors(ss.sparkContext).flatMap {
+ case (host, executors) =>
+ executors.map {
+ executor => s"executor_${host}_$executor"
+ }
+ }.toArray
override def internalCompute(split: Partition,
context: TaskContext): Iterator[String] = {
@@ -38,6 +43,14 @@ class InvalidateSegmentCacheRDD(@transient private val ss: SparkSession, carbonT
Iterator.empty
}
+ override protected def getPreferredLocations(split: Partition): Seq[String] = {
+ if (split.asInstanceOf[DataMapRDDPartition].getLocations != null) {
+ split.asInstanceOf[DataMapRDDPartition].getLocations.toSeq
+ } else {
+ Seq()
+ }
+ }
+
override protected def internalGetPartitions: Array[Partition] = {
if (invalidSegmentIds.isEmpty) {
Array()