Spark Scalaでgroupbyを使用せずにシーケンスIDを生成する必要があります

Question

テーブルのProduct_IDsが変更されると、Sequence of Numbers列（Seq_No）を生成します。私の入力テーブルでは、Product_IDsのみがあり、Seq_Noで出力します。 Scalaではサポートしていないため、GropuByや行番号をSQLのパーティションに使用することはできません。

Logic : Seq_No = 1 
      for(i = 2:No_of_Rows) 
      when Product_IDs(i) != Product_IDs(i-1) then Seq_No(i) = Seq_No(i-1)+1 
      Else Seq_No(i) = Seq_No(i-1) 
      end as Seq_No 

Product_IDs Seq_No 
ID1   1 
ID1   1 
ID1   1 
ID2   2 
ID3   3 
ID3   3 
ID3   3 
ID3   3 
ID1   4 
ID1   4 
ID4   5 
ID5   6 
ID3   7 
ID6   8 
ID6   8 
ID5   9 
ID5   9 
ID4   10 

だから私は、現在のPRODUCT_IDが前Product_Idsと等しくないとしてSEQ_NOを生成したいです。入力テーブルにはProduct_IDという列が1つしかありません.Spark Scalaを使用してProduct_IDとSeq_Noを欲しいとします。

Answer 1

私はおそらくちょうどシーケンス番号を生成する関数を記述します。

scala> val getSeqNum: String => Int = { 
    var prevId = "" 
    var n = 0 
    (id: String) => { 
    if (id != prevId) { 
     n += 1 
     prevId = id 
    } 
    n 
    } 
} 
getSeqNum: String => Int = <function1> 

scala> for { id <- Seq("foo", "foo", "bar") } yield getSeqNum(id) 
res8: Seq[Int] = List(1, 1, 2) 

UPDATE：私はあなたがその、ニキルを超えて欲しいもので、非常に明確ではない、と私はないです

スパークの専門家が、私はあなたがアルベルトBonsanto @

val rrd = ??? // Hopefully you know how to get the RRD 
for { 
    (id, col2, col3) <- rrd // assuming the entries are tuples 
    seqNum = getSeqNum(id) 
} yield ??? // Hopefully you know how to transform the entries