グループ化されたSpark RDDのコンテンツを個々の行としてフラット化してファイルに保存する方法

Question

キー（インデックス：Int）でグループ化されたRDD [（Int、Iterable [Coordinates]）]があります。座標はメンバを持つクラスである：

latitude: Double, longitude: Double

Iプリントを作成するか、または次のフォーム（各データポイントの行）であろうCSVファイルを作成したい：

と

index,latitude,longitude 

非グループ化されたRDD [（INT、座標）]それは、このように働いた：私はこのケースでそれを行うために管理するにはどうすればよい

val textOutputRDD = initialRDD.map(
    f => f._1.toString() + "," + f._2.latitude.toString() + "," + f._2.longitude.toString()) 
textOutputRDD.saveAsTextFile("TextOutput") 

？

Answer 1

単純なネストループが行います。ここでダブルスの単純な対でI近似座標：

val rdd = 
    sc.parallelize(
    Seq(
     1 -> Seq((4.1, 3.4), (5.6, 6.7), (3.4, 9.0)), 
     2 -> Seq((0.4, -4.1), (-3.4, 6.7), (7.0, 8.9)) 
    ) 
) 

val csvLike = 
    for ((key, coords) <- rdd; (lat, lon) <- coords) yield s"$key,$lat,$lon" 

for (row <- csvLike) println(row) 

このコードは次のような出力になります：

2,0.4,-4.1 
2,-3.4,6.7 
2,7.0,8.9 
1,4.1,3.4 
1,5.6,6.7 
1,3.4,9.0 

編集

別の可能なアプローチは/実際のflatMapに交換することですmapシーケンスの場合、コンパイラはforを次のように解釈します。

rdd.flatMap { 
    case (key, coords) => 
    coords.map { 
     case (lat, lon) => s"$key,$lat,$lon" 
    } 
} 

Answer 2

あなたがApacheのスパークのどのバージョンを使用しているflatmap-

val output = rdd.flatMap(s=>{ 
     val list=List[String]() 
     for (latlon <- s._2) { 
     list.addString(s._1.toString() + "," + latlon.latitude.toString() + "," + latlon.longitude.toString()) 
     } 
     return list 
    }) 
output.save(....)