Spark RDD を使用して SQL の「row_number()」関数を複製するにはどうすればよいですか?

Spark RDD 用に SQL の row_number() をレプリケートする方法?

SQL では、 row_number() 関数はテーブル内の各レコードに一意の行番号を割り当てます。指定された列のセットによってパーティション化され、別のセットによって順序付けされます。 Spark RDD のこの機能をレプリケートするには、次の手順を使用します。

1.キーと値のペアのPairRDDを作成します

タプル (K, V) のRDDを(K, (V, 1)) ペアのPairRDDに変換します。これにより、キー K とプレースホルダー値 1.

2 の両方に基づいて並べ替えを適用できるようになります。 RDD を並べ替えます

sortByKey() を使用して、キー K に基づいて、PairRDD を並べ替えます。これにより、同じキーを持つすべてのレコードがグループ化されます。

3. Zip With Index

ソートされた RDD で zipWithIndex() を呼び出して、各キー パーティション内の行番号を表す追加の列を追加します。

4.キー、値、および行番号を結合する

マッピング関数を使用して、PairRDD からキー、値、および行番号を抽出し、それらを新しいタプル (K、V、rowNum) に結合します。

このプロセスにより、RDD 内の各レコードに行番号を割り当てることができ、SQL の row_number() 関数の機能を効果的に複製できます。

例:

val temp1 = sc.parallelize(Seq(((3,4),5,5,5),((3,4),5,5,9),((3,4),7,5,5),((1,2),1,2,3),((1,2),1,4,7),((1,2),2,2,3)))

val temp2 = temp1
  .map(x => (x, 1))
  .sortByKey()
  .zipWithIndex()
  .map(a => (a._1._1, a._1._2._1, a._1._2._2, a._1._2._3, a._2 + 1))

temp2.collect().foreach(println)

出力:

((1,2),1,2,3,2)
((1,2),1,4,7,1)
((1,2),2,2,3,3)
((3,4),5,5,5,4)
((3,4),5,5,9,5)
((3,4),7,5,5,6)

以上がSpark RDD を使用して SQL の「row_number()」関数を複製するにはどうすればよいですか?の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。