3795

Build # 3795

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/3c0c1970f10ac01630617864dc5553143a303c2a">3c0c1970f">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/3c0c1970f10ac01630617864dc5553143a303c2a">3c0c1970f</a><a href="https://github.com/apache/carbondata/commit/3c0c1970f10ac01630617864dc5553143a303c2a">&lt;a href=&quot;https://github.com/apache/carbondata/commit/3c0c1970f10ac01630617864dc5553143a303c2a&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/apache/carbondata/commit/3c0c1970f10ac01630617864dc5553143a303c2a&amp;quot;&amp;gt;3c0c1970f&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/3c0c1970f10ac01630617864dc5553143a303c2a&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/3c0c1970f10ac01630617864dc555314&amp;lt;/a&amp;gt;3a303</a>c2a&quot;&gt;&amp;quot;&amp;gt;Merge &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/4790f8073&quot;&gt;4790f80... (continued)

Run Details

64034 of 82759 relevant lines covered (77.37%)

1.01 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.49

/integration/spark-common/src/main/scala/org/apache/spark/sql/hive/DistributionUtil.scala

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.spark.sql.hive

import java.net.{InetAddress, InterfaceAddress, NetworkInterface}

import scala.collection.JavaConverters._
import scala.util.control.Breaks._

import org.apache.spark.SparkContext
import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.block.Distributable
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.processing.util.CarbonLoaderUtil

object DistributionUtil {
  @transient
  val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
  /*
   *  minimum required registered resource for starting block distribution
   */
  lazy val minRegisteredResourceRatio: Double = {
    val value: String = CarbonProperties.getInstance()
      .getProperty(CarbonCommonConstants.CARBON_SCHEDULER_MIN_REGISTERED_RESOURCES_RATIO,
        CarbonCommonConstants.CARBON_SCHEDULER_MIN_REGISTERED_RESOURCES_RATIO_DEFAULT)
    java.lang.Double.parseDouble(value)
  }

  /*
   * node registration wait time
   */
  lazy val dynamicAllocationSchTimeOut: Integer = {
    val value: String = CarbonProperties.getInstance()
      .getProperty(CarbonCommonConstants.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_TIMEOUT,
        CarbonCommonConstants.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_TIMEOUT_DEFAULT)
    // milli second
    java.lang.Integer.parseInt(value) * 1000
  }

  /*
   * This method will return the list of executers in the cluster.
   * For this we take the  memory status of all node with getExecutorMemoryStatus
   * and extract the keys. getExecutorMemoryStatus also returns the driver memory also
   * In client mode driver will run in the localhost
   * There can be executor spawn in same drive node. So we can remove first occurance of
   * localhost for retriving executor list
   */
  def getNodeList(sparkContext: SparkContext): Array[String] = {
    val arr = sparkContext.getExecutorMemoryStatus.map { kv =>
      kv._1.split(":")(0)
    }.toSeq
    val localhostIPs = getLocalhostIPs
    val selectedLocalIPList = localhostIPs.filter(arr.contains(_))

    val nodelist: List[String] = withoutDriverIP(arr.toList)(selectedLocalIPList.contains(_))
    val masterMode = sparkContext.getConf.get("spark.master")
    if (nodelist.nonEmpty) {
      // Specific for Yarn Mode
      if ("yarn-cluster".equals(masterMode) || "yarn-client".equals(masterMode)) {
        val nodeNames = nodelist.map { x =>
          val addr = InetAddress.getByName(x)
          addr.getHostName
        }
        nodeNames.toArray
      } else {
        // For Standalone cluster, node IPs will be returned.
        nodelist.toArray
      }
    } else {
      Seq(InetAddress.getLocalHost.getHostName).toArray
    }
  }

  def getExecutors(sparkContext: SparkContext): Map[String, Seq[String]] = {
    val bm = sparkContext.env.blockManager
    bm.master.getPeers(bm.blockManagerId)
      .groupBy(blockManagerId => blockManagerId.host).map {
      case (host, blockManagerIds) => (host, blockManagerIds.map(_.executorId))
    }
  }

  private def getLocalhostIPs = {
    val iface = NetworkInterface.getNetworkInterfaces
    var addresses: List[InterfaceAddress] = List.empty
    while (iface.hasMoreElements) {
      addresses = iface.nextElement().getInterfaceAddresses.asScala.toList ++ addresses
    }
    val inets = addresses.map(_.getAddress.getHostAddress)
    inets
  }

  /*
   * This method will remove the first occurance of any of the ips  mentioned in the predicate.
   * Eg: l = List(Master,slave1,Master,slave2,slave3) is the list of nodes where first Master is
   * the Driver  node.
   * this method withoutFirst (l)(x=> x == 'Master') will remove the first occurance of Master.
   * The resulting List containt List(slave1,Master,slave2,slave3)
   */
  def withoutDriverIP[A](xs: List[A])(p: A => Boolean): List[A] = {
    xs match {
      case x :: rest => if (p(x)) {
        rest
      } else {
        x :: withoutDriverIP(rest)(p)
      }
      case _ => Nil
    }
  }

  /**
   *
   * Checking if the existing executors is greater than configured executors, if yes
   * returning configured executors.
   *
   * @param blockList total number of blocks in the identified segments
   * @param sparkContext
   * @return
   */
  def ensureExecutorsAndGetNodeList(blockList: Seq[Distributable],
      sparkContext: SparkContext): Seq[String] = {
    val nodeMapping = CarbonLoaderUtil.nodeBlockMapping(blockList.asJava)
    ensureExecutorsByNumberAndGetNodeList(nodeMapping, blockList, sparkContext)
  }

  def ensureExecutorsByNumberAndGetNodeList(nodesOfData: Int,
      sparkContext: SparkContext): Seq[String] = {
    val confExecutors: Int = getConfiguredExecutors(sparkContext)
    LOGGER.info(s"Executors configured : $confExecutors")
    val requiredExecutors = if (nodesOfData < 1 || nodesOfData > confExecutors) {
      confExecutors
    } else {
      nodesOfData
    }
    // request for starting the number of required executors
    ensureExecutors(sparkContext, requiredExecutors)
    getDistinctNodesList(sparkContext, requiredExecutors)
  }

  /**
   * This method will ensure that the required/configured number of executors are requested
   * for processing the identified blocks
   *
   * @param nodeMapping
   * @param blockList
   * @param sparkContext
   * @return
   */
  private def ensureExecutorsByNumberAndGetNodeList(
      nodeMapping: java.util.Map[String, java.util.List[Distributable]],
      blockList: Seq[Distributable],
      sparkContext: SparkContext): Seq[String] = {
    val nodesOfData = nodeMapping.size()
    val confExecutors: Int = getConfiguredExecutors(sparkContext)
    LOGGER.info(s"Executors configured : $confExecutors")
    val requiredExecutors = if (nodesOfData < 1) {
      1
    } else if (nodesOfData > confExecutors) {
      confExecutors
    } else if (confExecutors > nodesOfData) {
      var totalExecutorsToBeRequested = nodesOfData
      // If total number of blocks are greater than the nodes identified then ensure
      // that the configured number of max executors can be opened based on the difference of
      // block list size and nodes identified
      if (blockList.size > nodesOfData) {
        // e.g 1. blockList size = 40, confExecutors = 6, then all 6 executors
        // need to be opened
        // 2. blockList size = 4, confExecutors = 6, then
        // total 4 executors need to be opened
        if (blockList.size > confExecutors) {
          totalExecutorsToBeRequested = confExecutors
        } else {
          totalExecutorsToBeRequested = blockList.size
        }
      }
      LOGGER.info(s"Total executors requested: $totalExecutorsToBeRequested")
      totalExecutorsToBeRequested
    } else {
      nodesOfData
    }
    // request for starting the number of required executors
    ensureExecutors(sparkContext, requiredExecutors, blockList.size)
    getDistinctNodesList(sparkContext, requiredExecutors)
  }

  /**
   * This method will return the configured executors
   *
   * @param sparkContext
   * @return
   */
  private def getConfiguredExecutors(sparkContext: SparkContext): Int = {
    var confExecutors: Int = 0
    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
  }

  /**
   * This method will return the distinct nodes list
   *
   * @param sparkContext
   * @param requiredExecutors
   * @return
   */
  private def getDistinctNodesList(sparkContext: SparkContext,
      requiredExecutors: Int): Seq[String] = {
    val startTime = System.currentTimeMillis()
    var nodes = DistributionUtil.getNodeList(sparkContext)
    // calculate the number of times loop has to run to check for starting
    // the requested number of executors
    val threadSleepTime =
    CarbonCommonConstants.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_THREAD_SLEEP_TIME
    val maxRetryCount = calculateMaxRetry
    var maxTimes = maxRetryCount
    breakable {
      while (nodes.length < requiredExecutors && maxTimes > 0) {
        Thread.sleep(threadSleepTime);
        nodes = DistributionUtil.getNodeList(sparkContext)
        maxTimes = maxTimes - 1;
        val resourceRatio = (nodes.length.toDouble / requiredExecutors)
        if (resourceRatio.compareTo(minRegisteredResourceRatio) >= 0) {
          break
        }
      }
    }
    val timDiff = System.currentTimeMillis() - startTime
    LOGGER.info(s"Total Time taken to ensure the required executors : $timDiff")
    LOGGER.info(s"Time elapsed to allocate the required executors: " +
      s"${(maxRetryCount - maxTimes) * threadSleepTime}")
    nodes.distinct.toSeq
  }

  /**
   * Requesting the extra executors other than the existing ones.
   *
   * @param sc                   sparkContext
   * @param requiredExecutors         required number of executors to be requested
   * @param localityAwareTasks   The number of pending tasks which is locality required
   * @param hostToLocalTaskCount A map to store hostname with its possible task number running on it
   * @return
   */
  def ensureExecutors(sc: SparkContext,
      requiredExecutors: Int,
      localityAwareTasks: Int = 0,
      hostToLocalTaskCount: Map[String, Int] = Map.empty): Boolean = {
    sc.schedulerBackend match {
      case b: CoarseGrainedSchedulerBackend =>
        if (requiredExecutors > 0) {
          LOGGER.info(s"Requesting total executors: $requiredExecutors")
          b.requestTotalExecutors(requiredExecutors, localityAwareTasks, hostToLocalTaskCount)
        }
        true
      case _ =>
        false
    }
  }

  /**
   * This method will calculate how many times a loop will run with an interval of given sleep
   * time to wait for requested executors to come up
    *
    * @return The max retry count
   */
  def calculateMaxRetry(): Int = {
    val remainder = dynamicAllocationSchTimeOut % CarbonCommonConstants
      .CARBON_DYNAMIC_ALLOCATION_SCHEDULER_THREAD_SLEEP_TIME
    val retryCount: Int = dynamicAllocationSchTimeOut / CarbonCommonConstants
      .CARBON_DYNAMIC_ALLOCATION_SCHEDULER_THREAD_SLEEP_TIME
    if (remainder > 0) {
      retryCount + 1
    } else {
      retryCount
    }
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.spark.sql.hive
19
20	import java.net.{InetAddress, InterfaceAddress, NetworkInterface}
21
22	import scala.collection.JavaConverters._
23	import scala.util.control.Breaks._
24
25	import org.apache.spark.SparkContext
26	import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
27
28	import org.apache.carbondata.common.logging.LogServiceFactory
29	import org.apache.carbondata.core.constants.CarbonCommonConstants
30	import org.apache.carbondata.core.datastore.block.Distributable
31	import org.apache.carbondata.core.util.CarbonProperties
32	import org.apache.carbondata.processing.util.CarbonLoaderUtil
33
34	object DistributionUtil {
35	@transient
36	val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	1✔
37	/*
38	* minimum required registered resource for starting block distribution
39	*/
40	lazy val minRegisteredResourceRatio: Double = {	×
41	val value: String = CarbonProperties.getInstance()	×
42	.getProperty(CarbonCommonConstants.CARBON_SCHEDULER_MIN_REGISTERED_RESOURCES_RATIO,	×
43	CarbonCommonConstants.CARBON_SCHEDULER_MIN_REGISTERED_RESOURCES_RATIO_DEFAULT)	×
44	java.lang.Double.parseDouble(value)	×
45	}
46
47	/*
48	* node registration wait time
49	*/
50	lazy val dynamicAllocationSchTimeOut: Integer = {	1✔
51	val value: String = CarbonProperties.getInstance()	1✔
52	.getProperty(CarbonCommonConstants.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_TIMEOUT,	1✔
53	CarbonCommonConstants.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_TIMEOUT_DEFAULT)	1✔
54	// milli second
55	java.lang.Integer.parseInt(value) * 1000	1✔
56	}
57
58	/*
59	* This method will return the list of executers in the cluster.
60	* For this we take the memory status of all node with getExecutorMemoryStatus
61	* and extract the keys. getExecutorMemoryStatus also returns the driver memory also
62	* In client mode driver will run in the localhost
63	* There can be executor spawn in same drive node. So we can remove first occurance of
64	* localhost for retriving executor list
65	*/
66	def getNodeList(sparkContext: SparkContext): Array[String] = {	1✔
67	val arr = sparkContext.getExecutorMemoryStatus.map { kv =>	1✔
68	kv._1.split(":")(0)	1✔
69	}.toSeq	1✔
70	val localhostIPs = getLocalhostIPs	1✔
71	val selectedLocalIPList = localhostIPs.filter(arr.contains(_))	1✔
72
73	val nodelist: List[String] = withoutDriverIP(arr.toList)(selectedLocalIPList.contains(_))	1✔
74	val masterMode = sparkContext.getConf.get("spark.master")	1✔
75	if (nodelist.nonEmpty) {	1✔
76	// Specific for Yarn Mode
77	if ("yarn-cluster".equals(masterMode) \|\| "yarn-client".equals(masterMode)) {	×
78	val nodeNames = nodelist.map { x =>	×
79	val addr = InetAddress.getByName(x)	×
80	addr.getHostName	×
81	}
82	nodeNames.toArray	×
83	} else {
84	// For Standalone cluster, node IPs will be returned.
85	nodelist.toArray	×
86	}
87	} else {
88	Seq(InetAddress.getLocalHost.getHostName).toArray	1✔
89	}
90	}
91
92	def getExecutors(sparkContext: SparkContext): Map[String, Seq[String]] = {
93	val bm = sparkContext.env.blockManager	×
94	bm.master.getPeers(bm.blockManagerId)	×
95	.groupBy(blockManagerId => blockManagerId.host).map {	×
96	case (host, blockManagerIds) => (host, blockManagerIds.map(_.executorId))	×
97	}
98	}
99
100	private def getLocalhostIPs = {
101	val iface = NetworkInterface.getNetworkInterfaces	1✔
102	var addresses: List[InterfaceAddress] = List.empty	1✔
103	while (iface.hasMoreElements) {	1✔
104	addresses = iface.nextElement().getInterfaceAddresses.asScala.toList ++ addresses	1✔
105	}
106	val inets = addresses.map(_.getAddress.getHostAddress)	1✔
107	inets	1✔
108	}
109
110	/*
111	* This method will remove the first occurance of any of the ips mentioned in the predicate.
112	* Eg: l = List(Master,slave1,Master,slave2,slave3) is the list of nodes where first Master is
113	* the Driver node.
114	* this method withoutFirst (l)(x=> x == 'Master') will remove the first occurance of Master.
115	* The resulting List containt List(slave1,Master,slave2,slave3)
116	*/
117	def withoutDriverIP[A](xs: List[A])(p: A => Boolean): List[A] = {
118	xs match {	1✔
119	case x :: rest => if (p(x)) {	1✔
120	rest	1✔
121	} else {
122	x :: withoutDriverIP(rest)(p)	×
123	}
124	case _ => Nil	×
125	}
126	}
127
128	/**
129	*
130	* Checking if the existing executors is greater than configured executors, if yes
131	* returning configured executors.
132	*
133	* @param blockList total number of blocks in the identified segments
134	* @param sparkContext
135	* @return
136	*/
137	def ensureExecutorsAndGetNodeList(blockList: Seq[Distributable],
138	sparkContext: SparkContext): Seq[String] = {
139	val nodeMapping = CarbonLoaderUtil.nodeBlockMapping(blockList.asJava)	1✔
140	ensureExecutorsByNumberAndGetNodeList(nodeMapping, blockList, sparkContext)	1✔
141	}
142
143	def ensureExecutorsByNumberAndGetNodeList(nodesOfData: Int,
144	sparkContext: SparkContext): Seq[String] = {
145	val confExecutors: Int = getConfiguredExecutors(sparkContext)	1✔
146	LOGGER.info(s"Executors configured : $confExecutors")	1✔
147	val requiredExecutors = if (nodesOfData < 1 \|\| nodesOfData > confExecutors) {	1✔
148	confExecutors	1✔
149	} else {
150	nodesOfData	1✔
151	}
152	// request for starting the number of required executors
153	ensureExecutors(sparkContext, requiredExecutors)	1✔
154	getDistinctNodesList(sparkContext, requiredExecutors)	1✔
155	}
156
157	/**
158	* This method will ensure that the required/configured number of executors are requested
159	* for processing the identified blocks
160	*
161	* @param nodeMapping
162	* @param blockList
163	* @param sparkContext
164	* @return
165	*/
166	private def ensureExecutorsByNumberAndGetNodeList(
167	nodeMapping: java.util.Map[String, java.util.List[Distributable]],
168	blockList: Seq[Distributable],
169	sparkContext: SparkContext): Seq[String] = {
170	val nodesOfData = nodeMapping.size()	1✔
171	val confExecutors: Int = getConfiguredExecutors(sparkContext)	1✔
172	LOGGER.info(s"Executors configured : $confExecutors")	1✔
173	val requiredExecutors = if (nodesOfData < 1) {	1✔
174	1	×
175	} else if (nodesOfData > confExecutors) {	1✔
176	confExecutors	×
177	} else if (confExecutors > nodesOfData) {	1✔
178	var totalExecutorsToBeRequested = nodesOfData	×
179	// If total number of blocks are greater than the nodes identified then ensure
180	// that the configured number of max executors can be opened based on the difference of
181	// block list size and nodes identified
182	if (blockList.size > nodesOfData) {	×
183	// e.g 1. blockList size = 40, confExecutors = 6, then all 6 executors
184	// need to be opened
185	// 2. blockList size = 4, confExecutors = 6, then
186	// total 4 executors need to be opened
187	if (blockList.size > confExecutors) {	×
188	totalExecutorsToBeRequested = confExecutors	×
189	} else {
190	totalExecutorsToBeRequested = blockList.size	×
191	}
192	}
193	LOGGER.info(s"Total executors requested: $totalExecutorsToBeRequested")	×
194	totalExecutorsToBeRequested	×
195	} else {
196	nodesOfData	1✔
197	}
198	// request for starting the number of required executors
199	ensureExecutors(sparkContext, requiredExecutors, blockList.size)	1✔
200	getDistinctNodesList(sparkContext, requiredExecutors)	1✔
201	}
202
203	/**
204	* This method will return the configured executors
205	*
206	* @param sparkContext
207	* @return
208	*/
209	private def getConfiguredExecutors(sparkContext: SparkContext): Int = {
210	var confExecutors: Int = 0	1✔
211	if (sparkContext.getConf.getBoolean("spark.dynamicAllocation.enabled", false)) {	1✔
212	// default value for spark.dynamicAllocation.maxExecutors is infinity
213	confExecutors = sparkContext.getConf.getInt("spark.dynamicAllocation.maxExecutors", 1)	×
214	LOGGER.info(s"spark.dynamicAllocation.maxExecutors property is set to = $confExecutors")	×
215	} else {
216	// default value for spark.executor.instances is 2
217	confExecutors = sparkContext.getConf.getInt("spark.executor.instances", 1)	1✔
218	LOGGER.info(s"spark.executor.instances property is set to = $confExecutors")	1✔
219	}
220	confExecutors	1✔
221	}
222
223	/**
224	* This method will return the distinct nodes list
225	*
226	* @param sparkContext
227	* @param requiredExecutors
228	* @return
229	*/
230	private def getDistinctNodesList(sparkContext: SparkContext,
231	requiredExecutors: Int): Seq[String] = {
232	val startTime = System.currentTimeMillis()	1✔
233	var nodes = DistributionUtil.getNodeList(sparkContext)	1✔
234	// calculate the number of times loop has to run to check for starting
235	// the requested number of executors
236	val threadSleepTime =	1✔
237	CarbonCommonConstants.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_THREAD_SLEEP_TIME	1✔
238	val maxRetryCount = calculateMaxRetry	1✔
239	var maxTimes = maxRetryCount	1✔
240	breakable {	1✔
241	while (nodes.length < requiredExecutors && maxTimes > 0) {	1✔
242	Thread.sleep(threadSleepTime);	×
243	nodes = DistributionUtil.getNodeList(sparkContext)	×
244	maxTimes = maxTimes - 1;	×
245	val resourceRatio = (nodes.length.toDouble / requiredExecutors)	×
246	if (resourceRatio.compareTo(minRegisteredResourceRatio) >= 0) {	×
247	break	×
248	}
249	}
250	}
251	val timDiff = System.currentTimeMillis() - startTime	1✔
252	LOGGER.info(s"Total Time taken to ensure the required executors : $timDiff")	1✔
253	LOGGER.info(s"Time elapsed to allocate the required executors: " +	1✔
254	s"${(maxRetryCount - maxTimes) * threadSleepTime}")	1✔
255	nodes.distinct.toSeq	1✔
256	}
257
258	/**
259	* Requesting the extra executors other than the existing ones.
260	*
261	* @param sc sparkContext
262	* @param requiredExecutors required number of executors to be requested
263	* @param localityAwareTasks The number of pending tasks which is locality required
264	* @param hostToLocalTaskCount A map to store hostname with its possible task number running on it
265	* @return
266	*/
267	def ensureExecutors(sc: SparkContext,
268	requiredExecutors: Int,
269	localityAwareTasks: Int = 0,	1✔
270	hostToLocalTaskCount: Map[String, Int] = Map.empty): Boolean = {	1✔
271	sc.schedulerBackend match {	1✔
272	case b: CoarseGrainedSchedulerBackend =>	1✔
273	if (requiredExecutors > 0) {	×
274	LOGGER.info(s"Requesting total executors: $requiredExecutors")	×
275	b.requestTotalExecutors(requiredExecutors, localityAwareTasks, hostToLocalTaskCount)	×
276	}
277	true	×
278	case _ =>
279	false	1✔
280	}
281	}
282
283	/**
284	* This method will calculate how many times a loop will run with an interval of given sleep
285	* time to wait for requested executors to come up
286	*
287	* @return The max retry count
288	*/
289	def calculateMaxRetry(): Int = {	1✔
290	val remainder = dynamicAllocationSchTimeOut % CarbonCommonConstants	1✔
291	.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_THREAD_SLEEP_TIME	1✔
292	val retryCount: Int = dynamicAllocationSchTimeOut / CarbonCommonConstants	1✔
293	.CARBON_DYNAMIC_ALLOCATION_SCHEDULER_THREAD_SLEEP_TIME	1✔
294	if (remainder > 0) {	1✔
295	retryCount + 1	×
296	} else {
297	retryCount	1✔
298	}
299	}
300	}	1✔

apache / carbondata / 3795

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous