[scala] (왜) 캐시를 호출하거나 RDD를 유지해야합니까?

나는 그 질문이 다음과 같이 더 잘 공식화 될 것이라고 생각합니다.

언제 캐시를 호출하거나 RDD를 유지해야합니까?

스파크 프로세스는 게으르다. 즉, 필요할 때까지 아무 일도 일어나지 않을 것이다. 질문에 신속하게 답변하기 위해 val textFile = sc.textFile("/user/emp.txt")발행 된 후에 는 데이터가 아무 것도 발생하지 않으며 HadoopRDD파일을 소스로 사용하여 a 만 구성됩니다.

해당 데이터를 약간 변환한다고 가정 해 보겠습니다.

val wordsRDD = textFile.flatMap(line => line.split("\\W"))

다시 말하지만 데이터에는 아무런 변화가 없습니다. 이제 필요한 경우 적용 할 wordsRDD참조 testFile및 함수 가 포함 된 새로운 RDD 가 있습니다 .

과 같은 RDD에 대해 조치가 호출 될 때만 계보wordsRDD.count 라고하는 RDD 체인 이 실행됩니다. 즉, 파티션으로 분류 된 데이터는 Spark 클러스터의 실행 프로그램에 의해로드되고 함수가 적용되고 결과가 계산됩니다.flatMap

이 예의 선형 계보에서는 cache()필요하지 않습니다. 데이터가 실행기에로드되고 모든 변환이 적용되고 count데이터가 메모리에 맞는 경우 메모리에 모두 계산됩니다.

cacheRDD 계보가 분기 될 때 유용합니다. 이전 예제의 단어를 긍정적 인 단어와 부정적인 단어의 개수로 필터링한다고 가정 해 봅시다. 당신은 이렇게 할 수 있습니다 :

val positiveWordsCount = wordsRDD.filter(word => isPositive(word)).count()
val negativeWordsCount = wordsRDD.filter(word => isNegative(word)).count()

여기서 각 지점은 데이터를 다시로드합니다. 명시 적 cache명령문을 추가하면 이전에 수행 된 처리가 보존되고 재사용됩니다. 작업은 다음과 같습니다.

val textFile = sc.textFile("/user/emp.txt")
val wordsRDD = textFile.flatMap(line => line.split("\\W"))
wordsRDD.cache()
val positiveWordsCount = wordsRDD.filter(word => isPositive(word)).count()
val negativeWordsCount = wordsRDD.filter(word => isNegative(word)).count()

이러한 이유로 cache추가 처리를 위해 재사용 할 수있는 체크 포인트를 생성하므로 ‘계보를 깰’것으로 알려져 있습니다.

경험 법칙 : cacheRDD 계보가 분기 되거나 RDD가 루프 에서처럼 여러 번 사용될 때 사용합니다.

---

답변