Sparkでgroupbyの後にRDD操作を実行するにはどうすればよいですか？

Question

私は、クラスタリングを実行したい大量のデータを持っています。キャッチされている、私は全体のセットのための1つのクラスタリングを望んでいないが、各ユーザーのためのクラスタリングです。基本的に私はまずgroupbyのuseridを実行してから、KMeansを実行します。

問題は、いったんgroupbyを実行すると、すべてのマッピングがスパークコントローラのコンテキスト外にあるため、RDDを作成しようとしても失敗するということです。 mllibのSparkのKMeansのlibにはRDDが必要です（並列化できるように）。

私は2つの回避策を見ていますが、もっと良い解決策があることを期待していました。

1）コントローラー内の何千ものユーザーを（おそらく何かが大きくなると何百万もの）利用者に手動でループし、それぞれにkmeansを実行します。

2）コントローラーでgroupbyを実行し、マップ内で外部ライブラリーから提供されている非平行なkmeanを実行します。

別の方法があると教えてください。私はむしろすべてを持っています。できるだけ。

Answer 1

編集：応答の瞬間はpysparkだったことはわかりませんでした。しかし、私はそれを適応させる考えとして残しておきます。

私は同様の問題を抱えていましたが、パフォーマンスを改善することはできましたが、それでも私にとって理想的な解決策ではありませんでした。多分あなたのために働くかもしれません。

アイデアは、多くの小さなRDD（ユーザーIDごとに新しいもの）でRDDを分解し、アレイに保存してから、それぞれの "サブRDD"に対して処理機能（あなたのケースではクラスタリング）を呼び出していました。提案されたコードは（コメント欄で説明）以下の通りである：

// A case class just to use as example 
case class MyClass(userId: Long, value: Long, ...) 

// A Scala local array with the user IDs (Could be another iterator, such as List or Array): 
val userList: Seq[Long] = rdd.map{ _.userId }.distinct.collect.toSeq // Just a suggestion! 

// Now we can create the new rdds: 
val rddsList: Seq[RDD[MyClass]] = userList.map { 
    userId => rdd.filter({ item: MyClass => item.userId == userId }) 
}.toSeq 

// Finally, we call the function we want for each RDD, saving the results in a new list. 
// Note the ".par" call, which is used to start the expensive execution for multiple RDDs at the same time 
val results = rddsList.par.map { 
    r => myFunction(r) 
} 

私はこれが大体あなたの最初のオプションと同じですけど、.parコールを使用することによって、私はパフォーマンスを向上させることができました。

この呼び出しは、rddsListオブジェクトをParSeqオブジェクトに変換します。この新しいScalaオブジェクトは並列計算を可能にするので、マップ関数は複数のRDDに対して同時にmyFunction(r)を呼び出してパフォーマンスを向上させるのが理想的です。

並列コレクションの詳細については、Scala Documentationをご確認ください。