varでRDDをフィルタリングすると予期しない動作が発生する

Question

私のコードに奇妙なバグがあり、デバッグ中に問題を解決することができました。問題は、var RDDをvar変数でフィルタリングし、同じRDDにフィルタ結果を格納すると、RDDが正しく更新されることです。 問題は、var変数を更新した後、私は自動的に結果をフィルタリングするために使用されました再びフィルタリングすることです！

コード例：

var filter = 5 
var a1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9)) 
a1 = a1.filter(t => !t.equals(filter)) 
a1.foreach(println) // result is: 1-9 without 5 
filter = filter + 1 
a1.foreach(println) // result is: 1-9 without 6 

がなぜ起こっていますか？私のコードにバグを引き起こさないようにするためのルールは何ですか？

Answer 1

スパーク変換は遅延評価されます。 a1.filterを実行するとFilteredRDDが返されますが、その時点でという計算結果が実際にには含まれていません。 foreachを使用して変換要求とアクションを実行した場合のみ、変換が呼び出されます。

lazyフィルタリングと同様に、ラムダ式は変数をキャプチャしますが、値ではありません。つまり、filterを更新すると、キャプチャされたラムダ内の同じ変数が5から6に更新され、再度フィルタリングすると更新された値を持つすべての要素が生成されます。

Answer 2

これは、a1に完全なDAGが含まれているためです。 foreachはDAGをトリガして結果を得るアクションです。あなたはforeachのを使用してRDDを印刷するたび

scala> var a1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9)) 
a1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:21 

scala> a1.toDebugString 
res5: String = (4) ParallelCollectionRDD[4] at parallelize at <console>:21 [] 

scala> a1 = a1.filter(t => !t.equals(filter)) 
a1: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[5] at filter at <console>:25 


scala> a1.toDebugString 
res6: String = 
(4) MapPartitionsRDD[5] at filter at <console>:25 [] 
| ParallelCollectionRDD[4] at parallelize at <console>:21 [] 

だから、閉鎖にフィルタ値を取り、DAGを計算することによって、あなたに結果を取得します。

filter = 6 
a1.foreach(println) // will filter 6 
filter = 9 
a1.foreach(println) // will filter 9 

Answer 3

これらを試してみて、何が起こるかを参照してください。

var filter = 5 
var a1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9)) 
a1 = sc.parallelize(a1.filter(t => !t.equals(filter)).collect()) 
a1.foreach(println) 
filter = filter + 1 
a1.foreach(println) 

そして、これをも：

var filter = 5 
var a1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9)) 
a1 = a1.filter(t => !t.equals(filter)).cache() 
a1.foreach(println) 
filter = filter + 1 
a1.foreach(println) 

が、これらはあなたがより多くを考えさせることを願っています！