Wie teile ich Datensätze auf?

Question

Ich habe eine CSV-Datei und die Daten für customer_id sieht wie folgt aus.

CUSTID,LOC,PRODNAME,UNITS,TYPE,PURCHASE_DATE 
123,"SA","PROD1",1000,"PAY","20-DEC-2016" 
123,"SA","PROD2",500,"REC","31-AUG-2016" 

Und das hat umgewandelt in drei Datensätze ein als übergeordnete Datensatz und die beiden anderen als Kind Aufzeichnungen über die Spalte TYPE, wie unten zu stützen. Dies muss schließlich in eine andere CSV-Datei gehen.

ROWTYPE,CUSTID,LOC,TYPE,PRODUCT_NAME,UNITS,PURCHASE_DATE 
PARENT,123,"SA" 
CHILD,123,"SA","PAY","PROD1","20-DEC-2016" 
CHILD,123,"SA","REC","PROD2","31-AUG-2016" 

Zwei weitere Dinge auf die gleiche Aufgabe.

1. Meine Ausgabe CSV-Datei hat nur eine Kopfzeile nur für den übergeordneten Datensatz und untergeordnete Datensätze können mehr oder weniger Felder als der übergeordnete Datensatz haben. Aus diesem Grund kann ich UNION Eltern und Kinder nicht, also habe ich gerade alle Dataframes in RDD umgewandelt und eine Union auf diesem gemacht. Also nur zu wissen, wenn dies mit Dataframes getan werden muss, wie würde ich es tun?

2. Dann muss ich schließlich eine CSV-Datei in einer bestimmten Reihenfolge basierend auf CUSTID und TYPE-Feld erstellen. Ich weiß, dass dies einfach bei Datenrahmen ist, aber seit ich sie in RDDs umgewandelt habe, habe ich Folgendes getan, weiß aber nicht, ob es ein optimaler Ansatz ist.

    
   val rdd: RDD[((String,Int),Row)] = solution.map(row => ({ 
       val custId: String = row.getString(1) 
       val transType: Int = row.getString(0) match { 
       case "PAY" => 1 
       case "REC" => 2 
       case _ => 0 
       } 
       (custId,transType)},row)) 
   
   implicit val caseInsensitiveOrdering = new Ordering[String] { 
       override def compare(a: String, b: String) = a.toLowerCase().compare(b.toLowerCase()) 
   } 
   

Auch dies funktioniert nicht, wenn ich die Ausgabe CSV-Datei neu partitionieren und fusioniert, wie die Bestellung für einen Wurf geht.

Answer 1

Getting PARENT,123,"SA" scheint einfach durch eine einfache Abfrage mit groupBy.

val parents = customers. 
    select(lit("PARENT") as "ROWTYPE", $"custid", $"loc"). 
    dropDuplicates 
scala> parents.show 
+-------+------+---+ 
|ROWTYPE|custid|loc| 
+-------+------+---+ 
| PARENT| 123| SA| 
+-------+------+---+ 

Damit Sie union es mit dem Rest der endgültigen PARENT und CHILD Aufzeichnungen zu bekommen.

Da union kann nur mit der gleichen Anzahl von Spalten in Tabellen durchgeführt werden, und customers 6 Spalten und parents hat drei Spalten, müssen Sie die Datensätze übereinstimmen.

val fullParents = parents. 
    withColumn("PRODUCT_NAME", lit("")). 
    withColumn("UNITS", lit("")). 
    withColumn("TYPE", lit("")). 
    withColumn("PURCHASE_DATE", lit("")) 
scala> .show 
+-------+------+---+------------+-----+----+-------------+ 
|ROWTYPE|custid|loc|PRODUCT_NAME|UNITS|TYPE|PURCHASE_DATE| 
+-------+------+---+------------+-----+----+-------------+ 
| PARENT| 123| SA|   |  | |    | 
+-------+------+---+------------+-----+----+-------------+ 

Lassen Sie uns die ROWTYPE als CHILD zu customers hinzufügen.

val rowtypedCustomers = customers. 
    select(lit("CHILD") as "ROWTYPE", customers("*")). 
    withColumnRenamed("PRODNAME", "PRODUCT_NAME") 
scala> rowtypedCustomers.show 
+-------+------+---+------------+-----+----+-------------+ 
|ROWTYPE|CUSTID|LOC|PRODUCT_NAME|UNITS|TYPE|PURCHASE_DATE| 
+-------+------+---+------------+-----+----+-------------+ 
| CHILD| 123| SA|  PROD1| 1000| PAY| 20-DEC-2016| 
| CHILD| 123| SA|  PROD2| 500| REC| 31-AUG-2016| 
+-------+------+---+------------+-----+----+-------------+ 

val solution = fullParents.union(rowtypedCustomers) 
scala> .show 
+-------+------+---+------------+-----+----+-------------+ 
|ROWTYPE|custid|loc|PRODUCT_NAME|UNITS|TYPE|PURCHASE_DATE| 
+-------+------+---+------------+-----+----+-------------+ 
| PARENT| 123| SA|   |  | |    | 
| CHILD| 123| SA|  PROD1| 1000| PAY| 20-DEC-2016| 
| CHILD| 123| SA|  PROD2| 500| REC| 31-AUG-2016| 
+-------+------+---+------------+-----+----+-------------+ 

als CSV-Schreiben ist so einfach wie die folgende Abfrage:

solution.write.csv("solution.csv") 

Erledigt. Herzlichen Glückwunsch!