Как HDInsigh Spark работает со следующим кодом

Question

Я новичок в Spark и HDInsight.Мне интересно, как работает код ниже?Я имею в виду, как планируются работы среди узлов кластера.Предположим, у меня есть 2 головных узла и 4 рабочих узла, какие коды будут выполняться на головных узлах и какие коды будут выполняться на рабочих узлах?

Как я могу сообщить Spark, что я хочу, чтобы какой-то код выполнялся на рабочих узлах?

// the code is from https://docs.microsoft.com/en-us/azure/hdinsight/spark/apache-spark-eventhub-streaming
import com.microsoft.azure.eventhubs._
val pool = Executors.newFixedThreadPool(1)
val eventHubClient = EventHubClient.create(connStr.toString(), pool)

def sendEvent(message: String) = {
      val messageData = EventData.create(message.getBytes("UTF-8"))
      eventHubClient.get().send(messageData)
      println("Sent event: " + message + "\n")
}

import twitter4j._
import twitter4j.TwitterFactory
import twitter4j.Twitter
import twitter4j.conf.ConfigurationBuilder

// Twitter application configurations
// Replace values below with yours   
val twitterConsumerKey = "<CONSUMER KEY>"
val twitterConsumerSecret = "<CONSUMER SECRET>"
val twitterOauthAccessToken = "<ACCESS TOKEN>"
val twitterOauthTokenSecret = "<TOKEN SECRET>"

val cb = new ConfigurationBuilder()
cb.setDebugEnabled(true).setOAuthConsumerKey(twitterConsumerKey).setOAuthConsumerSecret(twitterConsumerSecret).setOAuthAccessToken(twitterOauthAccessToken).setOAuthAccessTokenSecret(twitterOauthTokenSecret)

val twitterFactory = new TwitterFactory(cb.build())
val twitter = twitterFactory.getInstance()

// Getting tweets with keyword "Azure" and sending them to the Event Hub in realtime!

val query = new Query(" #Azure ")
query.setCount(100)
query.lang("en")
var finished = false
while (!finished) {
  val result = twitter.search(query)
  val statuses = result.getTweets()
  var lowestStatusId = Long.MaxValue
  for (status <- statuses.asScala) {
    if(!status.isRetweet()){
      sendEvent(status.getText())
    }
    lowestStatusId = Math.min(status.getId(), lowestStatusId)
    Thread.sleep(2000)
  }
  query.setMaxId(lowestStatusId - 1)
}

// Closing connection to the Event Hub
eventHubClient.get().close()

Answer 1

Добро пожаловать в Stackoverflow и Azure!

Легко понять компоненты Spark, поняв, как Spark работает на кластерах HDInsight.

Приложения Spark запускаются как независимые наборы процессов в кластере, координируемые объектом SparkContext в вашей основной программе (называемой программой драйвера).

SparkContext может подключаться к нескольким типам диспетчеров кластеров, которые распределяют ресурсы между приложениями.К этим менеджерам кластеров относятся Apache Mesos, Apache Hadoop YARN или диспетчер кластеров Spark.В HDInsight Spark работает с помощью менеджера кластеров YARN.После подключения Spark получает исполнителей на рабочих узлах кластера, которые представляют собой процессы, которые выполняют вычисления и хранят данные для вашего приложения.Затем он отправляет исполнителям код вашего приложения (определенный с помощью файлов JAR или Python, передаваемых в SparkContext).Наконец, SparkContext отправляет задачи исполнителям для запуска.

SparkContext выполняет основную функцию пользователя и выполняет различные параллельные операции на рабочих узлах.Затем SparkContext собирает результаты операций.Рабочие узлы считывают и записывают данные из и в распределенную файловую систему Hadoop.Рабочие узлы также кэшируют преобразованные данные в памяти как устойчивые распределенные наборы данных (RDD).

SparkContext подключается к ведущему устройству Spark и отвечает за преобразование приложения в ориентированный граф (DAG) отдельных задач, которые получаютвыполняется в процессе исполнителя на рабочих узлах.Каждое приложение получает свои собственные процессы-исполнители, которые работают в течение всего приложения и выполняют задачи в нескольких потоках.

Надеюсь, это поможет.