Odtnhj

I have a CSV file and the data for customer_id looks like below.

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

And this has to be converted into three records one as a Parent record and other two as Child records based on the TYPE column as below. This has to finally go into another CSV file.

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"

Two more things on the same task.

1. My output CSV file has only one header just for the parent record and child records could have either more or less fields than the parent record.
   Because of this reason I cannot UNION parent and childs so I just converted all the Dataframes into RDD and did a union on that.
   So to just know, if this has to be done using Dataframes, how would I do it?




2. Then finally I need to create a CSV file in some specific order based on CUSTID and TYPE field. I know this is easy incase of Dataframes but since I converted them to RDD's I did the following but don't know if it's an optimum approach.

Even this one doesn't work when I repartition and merge the output CSV file as the ordering goes for a toss.

Getting PARENT,123,"SA" seems easy by a simple query with groupBy.

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

With that, you union it with the rest to get the final PARENT and CHILD records.

Since union can only be performed on tables with the same number of columns, and customers has 6 columns and parents has 3 columns, you have to make the datasets match.

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| | | | |
+-------+------+---+------------+-----+----+-------------+

Let's add the ROWTYPE as CHILD to customers.

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|
+-------+------+---+------------+-----+----+-------------+

Writing it as a CSV is as simple as the following query:

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

Done. Congrats!