dropDuplicatesに関する問題（）とスカラ座

Question

を使用してスパークで（）メソッドを除いて、私はcolum xcept（filteredDuplicates）を持つテーブルがある

私はスパークシェルでこれを実行すると、予想通り、それは完全に正常に動作しています。しかし、スパーク送信では、重複削除はソートされた順序ではありません（すなわち、seq_no 3は有効なフレームにあり、1,5は拒否されたフレームにあります）。また、except（）もspark submitで問題を抱えています。私はこれで一日中立ち往生しています。誰か助けてください 皆さんありがとうございます

Answer 1

次のメソッドは、spark-sqlが提供するrow_number関数を使用して、有効なデータフレームと無効なデータフレームを生成します。私はcassandraへのアクセスを持っていないので、ここでは単純なDataframeを使用しています。

import sqlContext.implicits._ 
val df = sc.parallelize(Seq(("a" -> 1), ("b" -> 2), ("c" -> 3), ("d" -> 4), ("a" -> 5), ("a" -> 6), ("c" -> 7), ("c" -> 8))).toDF("c1", "c2") 

df.registerTempTable("temp_table") 

val masterdf = sqlContext.sql("SELECT c1, c2, ROW_NUMBER() OVER(PARTITION BY c1 ORDER BY c2) as row_num FROM temp_table") 

masterdf.filter("row_num = 1").show() 
+---+---+-------+ 
| c1| c2|row_num| 
+---+---+-------+ 
| a| 1|  1| 
| b| 2|  1| 
| c| 3|  1| 
| d| 4|  1| 
+---+---+-------+ 


masterdf.filter("row_num > 1").show() 
+---+---+-------+ 
| c1| c2|row_num| 
+---+---+-------+ 
| a| 5|  2| 
| a| 6|  3| 
| c| 7|  2| 
| c| 8|  3| 
+---+---+-------+