किसी कार्यकर्ता नोड पर निष्पादक के लिए कोर की संख्या प्राप्त करने का तरीका?

Question

ईजी। मुझे सभी उपलब्ध निष्पादकों और उनकी संबंधित मल्टीथ्रेडिंग क्षमता की सूची प्राप्त करने की आवश्यकता है (कुल मल्टीथ्रेडिंग क्षमता नहीं, sc.defaultParallelism पहले से ही इसे संभालती है)।

चूंकि यह पैरामीटर कार्यान्वयन-निर्भर है (यार्न और स्पार्क-स्टैंडअलोन में कोर आवंटित करने के लिए अलग-अलग रणनीति है) और स्थितिगत (यह गतिशील आवंटन और दीर्घकालिक नौकरी चलाने के कारण उतार-चढ़ाव कर सकता है)। मैं इसका अनुमान लगाने के लिए अन्य विधि का उपयोग नहीं कर सकता। क्या वितरित परिवर्तन में स्पार्क एपीआई का उपयोग करके इस जानकारी को पुनर्प्राप्त करने का कोई तरीका है?

1) कई विभाजन (के साथ एक 1 चरण काम >> defaultParallelism चलाने) और की संख्या की गणना: (उदाहरण के लिए TaskContext, SparkEnv)

अद्यतन स्पार्क 1.6 का सवाल है, मैं निम्न विधियों की कोशिश की है प्रत्येक executorID के लिए विशिष्ट threadIDs:

val n = sc.defaultParallelism * 16 
sc.parallelize(n, n).map(v => SparkEnv.get.executorID -> Thread.currentThread().getID) 
.groupByKey() 
.mapValue(_.distinct) 
.collect() 

हालांकि यह एक अनुमान वास्तविक बहु सूत्रण क्षमता की तुलना में अधिक की ओर जाता है, क्योंकि प्रत्येक स्पार्क निष्पादक एक overprovisioned थ्रेड पूल का उपयोग करता है।

2) 1 के समान, n = defaultParallesim को छोड़कर, और प्रत्येक कार्य में मैं असंतुलित sharding से संसाधन वार्ताकार को रोकने के लिए देरी जोड़ता हूं (एक तेज़ नोड इसे पूरा करता है और धीमे नोड्स चलने से पहले और पूछता है):

val n = sc.defaultParallelism 
sc.parallelize(n, n).map{ 
    v => 
    Thread.sleep(5000) 
    SparkEnv.get.executorID -> Thread.currentThread().getID 
} 
.groupByKey() 
.mapValue(_.distinct) 
.collect() 

यह समय के सबसे अधिक काम करता है, लेकिन आवश्यक तुलना में बहुत धीमी है और बहुत असंतुलित क्लस्टर या कार्य अटकलों से तोड़ा जा सकता है।

3) मैंने यह कोशिश नहीं की है: BlockManager.numUsableCores पढ़ने के लिए जावा प्रतिबिंब का उपयोग करें, यह स्पष्ट रूप से एक स्थिर समाधान नहीं है, आंतरिक कार्यान्वयन किसी भी समय बदल सकता है।

कृपया मुझे बताएं कि क्या आपको कुछ बेहतर मिला है।

Answer 1

स्पार्क आराम API के साथ यह बहुत आसान है।

val applicationId = spark.sparkContext.applicationId 

ui यूआरएल::

val baseUrl = spark.sparkContext.uiWebUrl 

और क्वेरी:

val url = baseUrl.map { url => 
    s"${url}/api/v1/applications/${applicationId}/executors" 
} 

अपाचे HTTP पुस्तकालय (पहले से ही स्पार्क निर्भरता में, https://alvinalexander.com/scala/scala-rest-client-apache-httpclient-restful-clients से रूपांतरित) के साथ: आप आवेदन आईडी प्राप्त करने के लिए

import org.apache.http.impl.client.DefaultHttpClient 
import org.apache.http.client.methods.HttpGet 
import scala.util.Try 

val client = new DefaultHttpClient() 

val response = url 
    .flatMap(url => Try{client.execute(new HttpGet(url))}.toOption) 
    .flatMap(response => Try{ 
    val s = response.getEntity().getContent() 
    val json = scala.io.Source.fromInputStream(s).getLines.mkString 
    s.close 
    json 
    }.toOption) 

और json4s:

import org.json4s._ 
import org.json4s.jackson.JsonMethods._ 
implicit val formats = DefaultFormats 

case class ExecutorInfo(hostPort: String, totalCores: Int) 

val executors: Option[List[ExecutorInfo]] = response.flatMap(json => Try { 
    parse(json).extract[List[ExecutorInfo]] 
}.toOption) 

जब तक आप हाथ और बाहरी कनेक्शन आप किसी भी कार्य से एक ही बात कर सकते हैं के लिए खुला ui बंदरगाह पर आवेदन आईडी और ui यूआरएल रहते हैं।

Answer 2

मैं वेब UI के समान तरीके से SparkListener को लागू करने का प्रयास करूंगा। उदाहरण के रूप में This code सहायक हो सकता है।