[SCALA] 스칼라 스파크, listbuffer가 비어 있습니다

스칼라 스파크, listbuffer가 비어 있습니다

listbuffer 항목의 주석 1 길이의 코드이 작품에서 제대로 표시되지만 2 주석 코드를 실행하지 않습니다. 왜 발생합니까?

val conf = new SparkConf().setAppName("app").setMaster("local")
val sc = new SparkContext(conf)

var wktReader: WKTReader = new WKTReader(); 
val dataSet = sc.textFile("dataSet.txt")

val items = new ListBuffer[String]() 
dataSet.foreach { e =>
  items += e
  println("len = " + items.length) //1. here length is ok
}

println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
items.foreach { x => print(x)} //2. this code doesn't be executed

로그는 여기에 있습니다 :

16/11/20 01:16:52 INFO Utils: Successfully started service 'SparkUI' on port 4040.
    16/11/20 01:16:52 INFO SparkUI: Bound SparkUI to 0.0.0.0, and started at http://192.168.56.1:4040
    16/11/20 01:16:53 INFO Executor: Starting executor ID driver on host localhost
    16/11/20 01:16:53 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 58608.
    16/11/20 01:16:53 INFO NettyBlockTransferService: Server created on 192.168.56.1:58608
    16/11/20 01:16:53 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, 192.168.56.1, 58608)
    16/11/20 01:16:53 INFO BlockManagerMasterEndpoint: Registering block manager 192.168.56.1:58608 with 347.1 MB RAM, BlockManagerId(driver, 192.168.56.1, 58608)
    16/11/20 01:16:53 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, 192.168.56.1, 58608)
    Starting app
    16/11/20 01:16:57 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 139.6 KB, free 347.0 MB)
    16/11/20 01:16:58 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 15.9 KB, free 346.9 MB)
    16/11/20 01:16:58 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on 192.168.56.1:58608 (size: 15.9 KB, free: 347.1 MB)
    16/11/20 01:16:58 INFO SparkContext: Created broadcast 0 from textFile at main.scala:25
    16/11/20 01:16:58 INFO FileInputFormat: Total input paths to process : 1
    16/11/20 01:16:58 INFO SparkContext: Starting job: foreach at main.scala:28
    16/11/20 01:16:58 INFO DAGScheduler: Got job 0 (foreach at main.scala:28) with 1 output partitions
    16/11/20 01:16:58 INFO DAGScheduler: Final stage: ResultStage 0 (foreach at main.scala:28)
    16/11/20 01:16:58 INFO DAGScheduler: Parents of final stage: List()
    16/11/20 01:16:58 INFO DAGScheduler: Missing parents: List()
    16/11/20 01:16:58 INFO DAGScheduler: Submitting ResultStage 0 (dataSet.txt MapPartitionsRDD[1] at textFile at main.scala:25), which has no missing parents
    16/11/20 01:16:58 INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 3.3 KB, free 346.9 MB)
    16/11/20 01:16:58 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 2034.0 B, free 346.9 MB)
    16/11/20 01:16:58 INFO BlockManagerInfo: Added broadcast_1_piece0 in memory on 192.168.56.1:58608 (size: 2034.0 B, free: 347.1 MB)
    16/11/20 01:16:58 INFO SparkContext: Created broadcast 1 from broadcast at DAGScheduler.scala:1012
    16/11/20 01:16:59 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 0 (dataSet.txt MapPartitionsRDD[1] at textFile at main.scala:25)
    16/11/20 01:16:59 INFO TaskSchedulerImpl: Adding task set 0.0 with 1 tasks
    16/11/20 01:16:59 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, localhost, partition 0, PROCESS_LOCAL, 5427 bytes)
    16/11/20 01:16:59 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
    16/11/20 01:16:59 INFO HadoopRDD: Input split: file:/D:/dataSet.txt:0+291
    16/11/20 01:16:59 INFO deprecation: mapred.tip.id is deprecated. Instead, use mapreduce.task.id
    16/11/20 01:16:59 INFO deprecation: mapred.task.id is deprecated. Instead, use mapreduce.task.attempt.id
    16/11/20 01:16:59 INFO deprecation: mapred.task.is.map is deprecated. Instead, use mapreduce.task.ismap
    16/11/20 01:16:59 INFO deprecation: mapred.task.partition is deprecated. Instead, use mapreduce.task.partition
    16/11/20 01:16:59 INFO deprecation: mapred.job.id is deprecated. Instead, use mapreduce.job.id
    len = 1
    len = 2
    len = 3
    len = 4
    len = 5
    len = 6
    len = 7
    16/11/20 01:16:59 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 989 bytes result sent to driver
    16/11/20 01:16:59 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 417 ms on localhost (1/1)
    16/11/20 01:16:59 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool 
    16/11/20 01:16:59 INFO DAGScheduler: ResultStage 0 (foreach at main.scala:28) finished in 0,456 s
    16/11/20 01:16:59 INFO DAGScheduler: Job 0 finished: foreach at main.scala:28, took 0,795126 s
    !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
    16/11/20 01:16:59 INFO SparkContext: Invoking stop() from shutdown hook
    16/11/20 01:16:59 INFO SparkUI: Stopped Spark web UI at http://192.168.56.1:4040
    16/11/20 01:16:59 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
    16/11/20 01:16:59 INFO MemoryStore: MemoryStore cleared
    16/11/20 01:16:59 INFO BlockManager: BlockManager stopped
    16/11/20 01:16:59 INFO BlockManagerMaster: BlockManagerMaster stopped
    16/11/20 01:16:59 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
    16/11/20 01:16:59 INFO SparkContext: Successfully stopped SparkContext
    16/11/20 01:16:59 INFO ShutdownHookManager: Shutdown hook called
    16/11/20 01:16:59 INFO ShutdownHookManager: Deleting directory

해결법

1. ==============================

   1.아파치 스파크 그러므로 여기에 공유 메모리를 제공하지 않습니다

   아파치 스파크 그러므로 여기에 공유 메모리를 제공하지 않습니다

   dataSet.foreach { e =>
     items += e
     println("len = " + items.length) //1. here length is ok
   }
   

   당신은 각각의 집행에 항목의 로컬 복사본을 수정합니다. 드라이버에 정의 된 원래의 항목 목록은 수정되지 않습니다. 그 결과이 :

   items.foreach { x => print(x) }
   

   실행하지만 인쇄 할 아무것도 없다.

   이해 폐쇄를 확인하시기 바랍니다

   그것은 여기에 추천 될 수 있지만, 당신은 누적있는 항목을 대체 할 수

   val acc = sc.collectionAccumulator[String]("Items")
   dataSet.foreach(e => acc.add(e))
   

2. ==============================

   2.스파크는 executers에서 실행하고 결과를 반환합니다. 의도 한대로 위의 코드는 작동하지 않습니다. 당신이 foreach 문에서 요소를 추가해야하는 경우 다음 드라이버에서 데이터를 수집하고 current_set에 추가해야합니다. 그러나 데이터를 수집하면 큰 데이터가 나쁜 생각이다.

   스파크는 executers에서 실행하고 결과를 반환합니다. 의도 한대로 위의 코드는 작동하지 않습니다. 당신이 foreach 문에서 요소를 추가해야하는 경우 다음 드라이버에서 데이터를 수집하고 current_set에 추가해야합니다. 그러나 데이터를 수집하면 큰 데이터가 나쁜 생각이다.

   val items = new ListBuffer[String]()
   
   val rdd = spark.sparkContext.parallelize(1 to 10, 4)
   rdd.collect().foreach(data => items += data.toString())
   println(items)
   

   산출:

   ListBuffer(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
   

from https://stackoverflow.com/questions/40699432/scala-spark-listbuffer-is-empty by cc-by-sa and MIT license