Repository: carbondata Updated Branches: refs/heads/master a5f080b67 -> eeeaf50f1
[CARBONDATA-3132]Correct the task disrtibution in case of compaction when the actual block nodes and active nodes are different Why This PR? There is an unequal distribution of tasks during compaction ex: When the load is done using replication factor 2 and all nodes are active and during compaction one node is down, basically it is not active executor, so the task distribution should take care to distribute the tasks equally among all the active executors instead of giving more tasks to single executor and less to other executor. But sometimes the unequal distribution happens and the compaction becomes sow. Solution Currently we are not getting active executors before the node block mapping and sending the list of active executors as null, which will lead to the above problem sometimes. so get the active executors and send for node block mapping logic which will handle to distribute equally. This closes #2953 Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/eeeaf50f Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/eeeaf50f Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/eeeaf50f Branch: refs/heads/master Commit: eeeaf50f120faa542635b94acb4a9d71ea203970 Parents: a5f080b Author: akashrn5 <[email protected]> Authored: Tue Nov 27 13:32:05 2018 +0530 Committer: ravipesala <[email protected]> Committed: Wed Nov 28 15:47:13 2018 +0530 ---------------------------------------------------------------------- .../spark/rdd/CarbonIUDMergerRDD.scala | 6 ++--- .../carbondata/spark/rdd/CarbonMergerRDD.scala | 28 ++++++-------------- .../spark/sql/hive/DistributionUtil.scala | 2 ++ .../spark/rdd/CarbonTableCompactor.scala | 19 ++----------- .../processing/util/CarbonLoaderUtil.java | 10 ++++--- .../apache/carbondata/tool/CarbonCliTest.java | 3 ++- 6 files changed, 22 insertions(+), 46 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/carbondata/blob/eeeaf50f/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonIUDMergerRDD.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonIUDMergerRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonIUDMergerRDD.scala index 8e49fab..9b4e3f6 100644 --- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonIUDMergerRDD.scala +++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonIUDMergerRDD.scala @@ -41,13 +41,11 @@ class CarbonIUDMergerRDD[K, V]( @transient private val ss: SparkSession, result: MergeResult[K, V], carbonLoadModel: CarbonLoadModel, - carbonMergerMapping: CarbonMergerMapping, - confExecutorsTemp: String) + carbonMergerMapping: CarbonMergerMapping) extends CarbonMergerRDD[K, V](ss, result, carbonLoadModel, - carbonMergerMapping, - confExecutorsTemp) { + carbonMergerMapping) { override def internalGetPartitions: Array[Partition] = { val startTime = System.currentTimeMillis() http://git-wip-us.apache.org/repos/asf/carbondata/blob/eeeaf50f/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala index 1fbcc51..5bcc49b 100644 --- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala +++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala @@ -64,8 +64,7 @@ class CarbonMergerRDD[K, V]( @transient private val ss: SparkSession, result: MergeResult[K, V], carbonLoadModel: CarbonLoadModel, - carbonMergerMapping: CarbonMergerMapping, - confExecutorsTemp: String) + carbonMergerMapping: CarbonMergerMapping) extends CarbonRDD[(K, V)](ss, Nil) { ss.sparkContext.setLocalProperty("spark.scheduler.pool", "DDL") @@ -401,26 +400,15 @@ class CarbonMergerRDD[K, V]( .add(new CarbonInputSplitTaskInfo(entry._1, entry._2).asInstanceOf[Distributable]) ) - val nodeBlockMap = CarbonLoaderUtil.nodeBlockMapping(taskInfoList, -1) + // get all the active nodes of cluster and prepare the nodeBlockMap based on these nodes + val activeNodes = DistributionUtil + .ensureExecutorsAndGetNodeList(taskInfoList.asScala, sparkContext) - val nodeTaskBlocksMap = new java.util.HashMap[String, java.util.List[NodeInfo]]() + val nodeBlockMap = CarbonLoaderUtil.nodeBlockMapping(taskInfoList, -1, activeNodes.asJava) - val confExecutors = confExecutorsTemp.toInt - val requiredExecutors = if (nodeBlockMap.size > confExecutors) { - confExecutors - } else { nodeBlockMap.size() } - DistributionUtil.ensureExecutors(sparkContext, requiredExecutors, taskInfoList.size) - logInfo("No.of Executors required=" + requiredExecutors + - " , spark.executor.instances=" + confExecutors + - ", no.of.nodes where data present=" + nodeBlockMap.size()) - var nodes = DistributionUtil.getNodeList(sparkContext) - var maxTimes = 30 - while (nodes.length < requiredExecutors && maxTimes > 0) { - Thread.sleep(500) - nodes = DistributionUtil.getNodeList(sparkContext) - maxTimes = maxTimes - 1 - } - logInfo("Time taken to wait for executor allocation is =" + ((30 - maxTimes) * 500) + "millis") + val nodeTaskBlocksMap = new java.util.HashMap[String, java.util.List[NodeInfo]]() + val nodes = DistributionUtil.getNodeList(sparkContext) + logInfo("no.of.nodes where data present=" + nodeBlockMap.size()) defaultParallelism = sparkContext.defaultParallelism val isPartitionTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable.isPartitionTable http://git-wip-us.apache.org/repos/asf/carbondata/blob/eeeaf50f/integration/spark-common/src/main/scala/org/apache/spark/sql/hive/DistributionUtil.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/hive/DistributionUtil.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/hive/DistributionUtil.scala index a676dd8..656d8eb 100644 --- a/integration/spark-common/src/main/scala/org/apache/spark/sql/hive/DistributionUtil.scala +++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/hive/DistributionUtil.scala @@ -203,9 +203,11 @@ object DistributionUtil { if (sparkContext.getConf.getBoolean("spark.dynamicAllocation.enabled", false)) { // default value for spark.dynamicAllocation.maxExecutors is infinity confExecutors = sparkContext.getConf.getInt("spark.dynamicAllocation.maxExecutors", 1) + LOGGER.info(s"spark.dynamicAllocation.maxExecutors property is set to = $confExecutors") } else { // default value for spark.executor.instances is 2 confExecutors = sparkContext.getConf.getInt("spark.executor.instances", 1) + LOGGER.info(s"spark.executor.instances property is set to = $confExecutors") } confExecutors } http://git-wip-us.apache.org/repos/asf/carbondata/blob/eeeaf50f/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala index 16309fe..cfb40ec 100644 --- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala +++ b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala @@ -168,19 +168,6 @@ class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel, OperationListenerBus.getInstance().fireEvent(dataMapPreExecutionEvent, dataMapOperationContext) } - var execInstance = "1" - // in case of non dynamic executor allocation, number of executors are fixed. - if (sc.sparkContext.getConf.contains("spark.executor.instances")) { - execInstance = sc.sparkContext.getConf.get("spark.executor.instances") - LOGGER.info(s"spark.executor.instances property is set to = $execInstance") - } // in case of dynamic executor allocation, taking the max executors of the dynamic allocation. - else if (sc.sparkContext.getConf.contains("spark.dynamicAllocation.enabled")) { - if (sc.sparkContext.getConf.get("spark.dynamicAllocation.enabled").trim - .equalsIgnoreCase("true")) { - execInstance = sc.sparkContext.getConf.get("spark.dynamicAllocation.maxExecutors") - LOGGER.info(s"spark.dynamicAllocation.maxExecutors property is set to = $execInstance") - } - } val mergeStatus = if (CompactionType.IUD_UPDDEL_DELTA == compactionType) { @@ -188,16 +175,14 @@ class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel, sc.sparkSession, new MergeResultImpl(), carbonLoadModel, - carbonMergerMapping, - execInstance + carbonMergerMapping ).collect } else { new CarbonMergerRDD( sc.sparkSession, new MergeResultImpl(), carbonLoadModel, - carbonMergerMapping, - execInstance + carbonMergerMapping ).collect } http://git-wip-us.apache.org/repos/asf/carbondata/blob/eeeaf50f/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java ---------------------------------------------------------------------- diff --git a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java index 3e4e31f..2563768 100644 --- a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java +++ b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java @@ -518,13 +518,14 @@ public final class CarbonLoaderUtil { /** * This method will divide the blocks among the nodes as per the data locality - * + * @param activeNodes List of all the active nodes running in cluster, based on these and the + * actual nodes, where blocks are present, the mapping is done * @param blockInfos * @return */ public static Map<String, List<Distributable>> nodeBlockMapping(List<Distributable> blockInfos, - int noOfNodesInput) { - return nodeBlockMapping(blockInfos, noOfNodesInput, null, + int noOfNodesInput, List<String> activeNodes) { + return nodeBlockMapping(blockInfos, noOfNodesInput, activeNodes, BlockAssignmentStrategy.BLOCK_NUM_FIRST,null); } @@ -536,7 +537,8 @@ public final class CarbonLoaderUtil { */ public static Map<String, List<Distributable>> nodeBlockMapping(List<Distributable> blockInfos) { // -1 if number of nodes has to be decided based on block location information - return nodeBlockMapping(blockInfos, -1); + return nodeBlockMapping(blockInfos, -1, null, + BlockAssignmentStrategy.BLOCK_NUM_FIRST,null); } /** http://git-wip-us.apache.org/repos/asf/carbondata/blob/eeeaf50f/tools/cli/src/test/java/org/apache/carbondata/tool/CarbonCliTest.java ---------------------------------------------------------------------- diff --git a/tools/cli/src/test/java/org/apache/carbondata/tool/CarbonCliTest.java b/tools/cli/src/test/java/org/apache/carbondata/tool/CarbonCliTest.java index e929b50..0c4158a 100644 --- a/tools/cli/src/test/java/org/apache/carbondata/tool/CarbonCliTest.java +++ b/tools/cli/src/test/java/org/apache/carbondata/tool/CarbonCliTest.java @@ -22,6 +22,7 @@ import java.io.File; import java.io.IOException; import java.io.PrintStream; +import org.apache.carbondata.core.constants.CarbonVersionConstants; import org.apache.carbondata.core.metadata.datatype.DataTypes; import org.apache.carbondata.core.util.CarbonUtil; import org.apache.carbondata.sdk.file.Field; @@ -205,7 +206,7 @@ public class CarbonCliTest { expectedOutput = buildLines( "## version Details", "written_by Version ", - "TestUtil 1.6.0-SNAPSHOT "); + "TestUtil "+ CarbonVersionConstants.CARBONDATA_VERSION+" "); Assert.assertTrue(output.contains(expectedOutput)); }
