Проблема с dropDuplicates() и, кроме() в Спарк с помощью Scala

Question

У меня есть таблица с Колум xcept (filteredDuplicates)

Когда я запускаю это в Спарк оболочки, он работает прекрасно, как и ожидалось. Но в иске submit, удаление дубликатов не находится в отсортированном порядке (т.е.) seq_no 3 находится в действительном кадре и 1,5 в отклоненном кадре. Кроме того, кроме() также возникает проблема в источнике submit. Я застрял в этом целый день, пожалуйста, помогите кому-нибудь помочь Спасибо за каждого заранее

Answer 1

Следующие методы генерируют действительные и недействительные данные, используя функцию row_number, предоставленную spark-sql. У меня нет доступа к 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| 
+---+---+-------+