Datenrahmen Transformationen mit verschachtelten Strukturen

Question

ich einen Funken Datenrahmen haben, die wie folgt aussieht:

root 
|-- employeeName: string (nullable = true) 
|-- employeeId: string (nullable = true) 
|-- employeeEmail: string (nullable = true) 
|-- company: struct (nullable = true) 
| |-- companyName: string (nullable = true) 
| |-- companyId: string (nullable = true) 
| |-- details: struct (nullable = true) 
| | |-- founded: string (nullable = true) 
| | |-- address: string (nullable = true) 
| | |-- industry: string (nullable = true) 

Was will ich tun Gruppe von CompanyID und eine Reihe von Mitarbeitern pro Unternehmen erhalten, wie folgt aus:

root 
|-- company: struct (nullable = true) 
| |-- companyName: string (nullable = true) 
| |-- companyId: string (nullable = true) 
| |-- details: struct (nullable = true) 
| | |-- founded: string (nullable = true) 
| | |-- address: string (nullable = true) 
| | |-- industry: string (nullable = true) 
|-- employees: array (nullable = true)  
| |-- employee: struct (nullable = true)   
| | |-- employeeName: string (nullable = true) 
| | |-- employeeId: string (nullable = true) 
| | |-- employeeEmail: string (nullable = true) 

Natürlich kann ich das leicht tun, wenn ich nur ein Paar (Firma, Mitarbeiter) hatte: (String, String) mit map und reduceByKey. Aber mit all den verschiedenen verschachtelten Informationen bin ich mir nicht sicher, welchen Ansatz ich wählen soll.

Sollte ich versuchen, alles zu glätten? Jedes Beispiel, um ähnliche Dinge zu tun, wäre sehr hilfreich.

Answer 1

Sie können tun, die folgenden -

// declaring data types 
case class Company(cName: String, cId: String, details: String) 
case class Employee(name: String, id: String, email: String, company: Company) 

// setting up example data 
val e1 = Employee("n1", "1", "n1@c1.com", Company("c1", "1", "d1")) 
val e2 = Employee("n2", "2", "n2@c1.com", Company("c1", "1", "d1")) 
val e3 = Employee("n3", "3", "n3@c1.com", Company("c1", "1", "d1")) 
val e4 = Employee("n4", "4", "n4@c2.com", Company("c2", "2", "d2")) 
val e5 = Employee("n5", "5", "n5@c2.com", Company("c2", "2", "d2")) 
val e6 = Employee("n6", "6", "n6@c2.com", Company("c2", "2", "d2")) 
val e7 = Employee("n7", "7", "n7@c3.com", Company("c3", "3", "d3")) 
val e8 = Employee("n8", "8", "n8@c3.com", Company("c3", "3", "d3")) 
val employees = Seq(e1, e2, e3, e4, e5, e6, e7, e8) 
val ds = sc.parallelize(employees).toDS 

// actual query to achieve what is mentioned in the question 
val result = ds.groupByKey(e => e.company).mapGroups((k, itr) => (k, itr.toList)) 
result.collect 

Ergebnisse in:

Array(

(Company(c1,1,d1),WrappedArray(Employee(n1,1,n1@c1.com,Company(c1,1,d1)), Employee(n2,2,n2@c1.com,Company(c1,1,d1)), Employee(n3,3,n3@c1.com,Company(c1,1,d1)))), 

(Company(c2,2,d2),WrappedArray(Employee(n4,4,n4@c2.com,Company(c2,2,d2)), Employee(n5,5,n5@c2.com,Company(c2,2,d2)), Employee(n6,6,n6@c2.com,Company(c2,2,d2)))), 

(Company(c3,3,d3),WrappedArray(Employee(n7,7,n7@c3.com,Company(c3,3,d3)), Employee(n8,8,n8@c3.com,Company(c3,3,d3))))) 

Das Wichtigste ist: Sie jede Funktion übergeben können Sie in mapGroups wollen die Gruppen in einer Art und Weise zu erhalten, Sie wollen.

Hoffe, das hilft.