Spark - 遍歷數據框中的所有行，比較每行的多列與另一列的比較

Question

| match_id | player_id | team | win | 
| 0  |  1 | A | A | 
| 0  |  2 | A | A | 
| 0  |  3 | B | A | 
| 0  |  4 | B | A | 
| 1  |  1 | A | B | 
| 1  |  4 | A | B | 
| 1  |  8 | B | B | 
| 1  |  9 | B | B | 
| 2  |  8 | A | A | 
| 2  |  4 | A | A | 
| 2  |  3 | B | A | 
| 2  |  2 | B | A | 

我有一個如上所示的數據幀。

我需要創建一個映射（鍵，值）對，從而使每一

(k=>(player_id_1, player_id_2), v=> 1)，如果player_id_1勝利反對player_id_2在比賽

和

(k=>(player_id_1, player_id_2), v=> 0)，如果player_id_1失去對player_id_2在比賽中

我將不得不重複遍歷整個數據幀，並根據其他3列將每個玩家ID與另一個進行比較。

我打算如下完成此操作。

1. 組由match_id

2. 在每個組中針對其他player_id一個player_id檢查是以下

   一個。如果match_id是相同的，並且團隊不同 然後

    if team = win 
        (k=>(player_id_1, player_id_2), v=> 0) 
       else team != win 
        (k=>(player_id_1, player_id_2), v=> 1) 
   

例如，通過匹配分區後考慮匹配1. player_id 1個需要進行比較，以player_id 2,3和4 雖然迭代， player_id 2的記錄將被跳過，因爲球隊是相同的 對於player_id 3，因爲球隊不同，球隊&贏得比較。 作爲player_id 1是A隊和player_id 3是B隊和A隊贏得形成將

((1,3),1) 

我如何在命令式編程實現這一公平理念的鍵值，但我真的很新的scala和函數式編程，並不能得知如何遍歷字段的每一行，如何通過檢查其他字段來創建（鍵，值）對。

我盡我所能解釋了這個問題。請讓我知道，如果我的問題的任何部分不清楚。我很樂意解釋這一點。謝謝。

PS：我使用的Spark 1.6

Answer 1

這可以通過使用數據幀API來實現如下圖所示..

數據幀API版本：

val df = Seq((0,1,"A","A"),(0,2,"A","A"),(0,3,"B","A"),(0,4,"B","A"),(1,1,"A","B"),(1,4,"A","B"),(1,8,"B","B"),(1,9,"B","B"),(2,8,"A","A"),(2,4,"A","A"),(2,3,"B","A"),(2,2,"B","A") 
).toDF("match_id", "player_id", "team", "win") 

val result = df.alias("left") 
     .join(df.alias("right"), $"left.match_id" === $"right.match_id" && not($"right.team" === $"left.team")) 
     .select($"left.player_id", $"right.player_id", when($"left.team" === $"left.win", 1).otherwise(0).alias("flag")) 

scala> result.collect().map(x => (x.getInt(0),x.getInt(1)) -> x.getInt(2)).toMap 
res4: scala.collection.immutable.Map[(Int, Int),Int] = Map((1,8) -> 0, (3,4) -> 0, (3,1) -> 0, (9,1) -> 1, (4,1) -> 0, (8,1) -> 1, (2,8) -> 0, (8,3) -> 1, (1,9) -> 0, (1,4) -> 1, (8,2) -> 1, (4,9) -> 0, (3,2) -> 0, (1,3) -> 1, (4,8) -> 0, (4,2) -> 1, (2,4) -> 1, (8,4) -> 1, (2,3) -> 1, (4,3) -> 1, (9,4) -> 1, (3,8) -> 0) 

SPARK SQL版本 ：

df.registerTempTable("data_table") 

val result = sqlContext.sql(""" 
SELECT DISTINCT t0.player_id, t1.player_id, CASE WHEN t0.team == t0.win THEN 1 ELSE 0 END AS flag FROM data_table t0 
INNER JOIN data_table t1 
ON t0.match_id = t1.match_id 
AND t0.team != t1.team 
""")