Spark에서 Executor Startup을위한 후크가 있습니까?

Question

기본적으로 공유 메모리에서 데이터를 읽으려면 동일한 노드/executor에서 실행되는 여러 태스크가 필요합니다. 이를 위해 작업을 시작하기 전에 데이터를 메모리에로드하는 초기화 함수가 필요합니다. Spark이 Executor 시작을위한 후크를 제공한다면이 초기화 코드를이 콜백 함수에 넣을 수 있습니다.이 초기화 코드는이 시작이 완료된 후에 만 ​​실행됩니다.

그래서 제 질문은 스파크가 그런 고리를 제공합니까? 그렇지 않은 경우 다른 방법을 사용하면 동일한 결과를 얻을 수 있습니까?

Answer 1

여러 개의 작업 (예 : 하나의 앱 인스턴스, 하나의 스파크 작업)을 실행할 수 있도록 앱의 여러 인스턴스를 실행할 필요는 없습니다. 동일한 SparkSession 객체를 여러 스레드가 사용하여 Spark 작업을 병렬로 제출할 수 있습니다.

그래서 다음과 같이 작동 할 수 있습니다

는

• 응용 프로그램이 시작되고 메모리에 공유 데이터를로드 할 초기화 함수를 실행합니다. 말해, SharedData 클래스 개체에.
• SparkSession 각 스레드 (SparkSession, SharedData)에 액세스 할
• 개체 각 스레드는 공유 SparkSession 및 SharedData 개체를 사용하여 점화 태스크를 생성
• 쓰레드 풀이 생성되어 생성된다.
• 사용 사례에 따라, 응용 프로그램은 다음 중 하나를 수행 : 모두 완료 할 작업에 대한
  • 대기를하고 새 요청이 도착하기를
  • 대기 루프에서 세션 불꽃 닫고 새로운 불꽃을 생성 필요에 따라 스레드 풀에서 스레드를 사용하여 작업. 사용 동시에 취소 할 수 있습니다 setJobGroup 때문에 관련 작업을 사용하여 작업에 그룹을 setJobDescription를 사용하여 작업 설명을 할당하거나 할당 같은 당 스레드 일을 할 때

SparkContext (sparkSession.sparkContext는) 유용 cancelJobGroup. 동일한 풀을 사용하는 작업에 대한 우선 순위를 조정할 수도 있습니다 (자세한 내용은 https://spark.apache.org/docs/latest/job-scheduling.html#scheduling-within-an-application 참조). 당신이 드라이버 응용 프로그램에서 한 번에 데이터를로드 어디 스파크를 직렬화하고 집행 (한 번)에 각각 전송 - "공유 데이터"에 대한

Answer 2

스파크의 솔루션은 방송을 사용하고 있습니다. 작업에서 해당 데이터를 사용하는 경우 Spark는 작업이 실행되기 전에 해당 데이터가 있는지 확인합니다. 예를 들면 : 당신이 드라이버 메모리에 데이터를 읽고 집행까지를 보내지 않도록하려면

object MySparkTransformation { 

    def transform(rdd: RDD[String], sc: SparkContext): RDD[Int] = { 
    val mySharedData: Map[String, Int] = loadDataOnce() 
    val broadcast = sc.broadcast(mySharedData) 
    rdd.map(r => broadcast.value(r)) 
    } 
} 

또는, 당신은 한 번씩 을 채워됩니다 값을 생성하기 위해 스칼라 object에 lazy 값을 사용할 수 있습니다 JVM, 이는 Spark의 경우 집행자 당 한 번입니다. 예를 들어 : 실제로

// must be an object, otherwise will be serialized and sent from driver 
object MySharedResource { 
    lazy val mySharedData: Map[String, Int] = loadDataOnce() 
} 

// If you use mySharedData in a Spark transformation, 
// the "local" copy in each executor will be used: 
object MySparkTransformation { 
    def transform(rdd: RDD[String]): RDD[Int] = { 
    // Spark won't include MySharedResource.mySharedData in the 
    // serialized task sent from driver, since it's "static" 
    rdd.map(r => MySharedResource.mySharedData(r)) 
    } 
} 

, 각 집행에 mySharedData의 복사본 하나를해야합니다.