스파크에서 saveAsTextFile 메서드

Question

내 프로젝트에서 나는 3 개의 입력 파일을 가지고 있으며 파일 이름을 args (0)로 args (2)로 만들고, 출력 파일 이름은 args (3)로, 소스 코드에서 로그에 내가 아무것도 할

val sc = new SparkContext() 
var log = sc.textFile(args(0)) 
for(i <- 1 until args.size - 1) log = log.union(sc.textFile(args(i))) 

사용하지만

log.coalesce(1, true).saveAsTextFile(args(args.size - 1)) 

를 사용하여 텍스트 파일로 저장하지만 여전히 일부-00001, 파트 00000으로 3 파일에 저장 부분-00002, 3 가지 입력 파일을 출력 파일에 저장할 수있는 방법이 있습니까?

Answer 1

여러 개의 출력 파일을 갖는 것은 Hadoop 또는 Spark와 같은 다중 컴퓨터 클러스터의 표준 동작입니다. 출력 파일의 수는 감속기의 수에 따라 다릅니다.

어떻게 하둡에서 그것을 "해결"하기 : merge output files after reduce phase

어떻게 스파크에서 "해결"하기 : how to make saveAsTextFile NOT split output into multiple file?

좋은 정보를 당신이 여기 얻을 수 있습니다 : http://apache-spark-user-list.1001560.n3.nabble.com/How-to-make-Spark-merge-the-output-file-td322.html

을 그래서, 당신은 바로 coalesce(1,true)에 관한 것이 었습니다. 그러나, 그것은 매우 비효율적이다. 흥미로운 점은 (@climbage에서 언급 한 것처럼) 로컬에서 실행하면 코드가 작동한다는 것입니다.

시도해 볼 수있는 것은 파일을 먼저 읽은 다음 출력을 저장하는 것입니다.

... 
val sc = new SparkContext() 
var str = new String("") 
for(i <- 0 until args.size - 1){ 
    val file = sc.textFile(args(i))  
    file.foreach(line => str+= line) 
} 
//and now you might save the content 
str.coalesce(1, true).saveAsTextFile("out") 

참고 :이 코드는 또한 매우 비효율적이며 작은 파일에만 사용할 수 있습니다. 더 나은 코드를 찾아야합니다. 나는 파일의 수를 줄이려고 시도하지 않고 여러 출력 파일을 대신 처리하려고한다.

Answer 2

앞서 언급했듯이 표준 API를 통해 귀하의 문제는 다소 피할 수 없는데, 그 이유는 귀하가 데이터의 대형 쿼티 (quanatities)를 다루고 있다는 가정 때문입니다. 나는 당신의 데이터를 관리 할 가정한다면, 당신은 시도 할 수있는 내가 여기서 뭐하는 거지 것은 수집하고 된 mkString을 수행하여 문자열로 RDD 변환됩니다

import java.nio.file.{Paths, Files}  
import java.nio.charset.StandardCharsets 

Files.write(Paths.get("./test_file"), data.collect.mkString("\n").getBytes(StandardCharsets.UTF_8)) 

다음과 같습니다. 나는 생산에서 이것을하지 않을 것을 제안 할 것이다. 로컬 데이터 분석을 위해 잘 작동합니다 (로컬 데이터의 5GB ~로 작업)