Chuyển đổi một khung dữ liệu Spark thành bộ sưu tập Bản đồ Scala

Question

Tôi đang cố tìm giải pháp tốt nhất để chuyển đổi toàn bộ khung dữ liệu Spark thành bộ sưu tập Bản đồ scala. Nó được minh họa tốt nhất như sau:

Để đi từ này (trong ví dụ Spark):

val df = sqlContext.read.json("examples/src/main/resources/people.json") 

df.show 
+----+-------+ 
| age| name| 
+----+-------+ 
|null|Michael| 
| 30| Andy| 
| 19| Justin| 
+----+-------+ 

Đối với một bộ sưu tập Scala (Bản đồ Maps) đại diện như thế này:

val people = Map(
Map("age" -> null, "name" -> "Michael"), 
Map("age" -> 30, "name" -> "Andy"), 
Map("age" -> 19, "name" -> "Justin") 
) 

Answer 1

tôi đừng nghĩ rằng câu hỏi của bạn có ý nghĩa - ngoài cùng của bạn Map, tôi chỉ thấy bạn đang cố gắng nhồi nhét các giá trị vào nó - bạn cần phải có cặp khóa/giá trị ở ngoài cùng của bạn Map. Điều đó đang được nói:

val peopleArray = df.collect.map(r => Map(df.columns.zip(r.toSeq):_*)) 

sẽ cung cấp cho bạn:

Array(
    Map("age" -> null, "name" -> "Michael"), 
    Map("age" -> 30, "name" -> "Andy"), 
    Map("age" -> 19, "name" -> "Justin") 
) 

Vào thời điểm đó bạn có thể làm:

val people = Map(peopleArray.map(p => (p.getOrElse("name", null), p)):_*) 

Trong đó sẽ cung cấp cho bạn:

Map(
    ("Michael" -> Map("age" -> null, "name" -> "Michael")), 
    ("Andy" -> Map("age" -> 30, "name" -> "Andy")), 
    ("Justin" -> Map("age" -> 19, "name" -> "Justin")) 
) 

tôi m đoán đây thực sự là những gì bạn muốn . Nếu bạn muốn để chìa khóa chúng trên một Long chỉ số tùy ý, bạn có thể làm:

val indexedPeople = Map(peopleArray.zipWithIndex.map(r => (r._2, r._1)):_*) 

nào mang đến cho bạn:

Map(
    (0 -> Map("age" -> null, "name" -> "Michael")), 
    (1 -> Map("age" -> 30, "name" -> "Andy")), 
    (2 -> Map("age" -> 19, "name" -> "Justin")) 
) 

Answer 2

tiên nhận được giản đồ từ Dataframe

val schemaList = dataframe.schema.map(_.name).zipWithIndex//get schema list from dataframe 

Lấy RDD từ dataframe và ánh xạ với nó

dataframe.rdd.map(row => 
    //here rec._1 is column name and rce._2 index 
    schemaList.map(rec => (rec._1, row(rec._2))).toMap 
).collect.foreach(println)