0
$\begingroup$

I've a data set with this fields:

Transaction_ID
Customer_ID
Department
Product_ID

And I'm trying to obtain a tuple with the products associated with each customer transaction. Like:

Transaction_1 -> Product_ID 1, Product_ID 2, Product_ID 3
Transaction_2 -> Product_ID 1, Product_ID 2, Product_ID 4
....

I've this code: But It not return the dataset as I want:

case class transactions (Transaction_ID: String, Customer_ID: String, Department: String, Product_ID: String)

def csvToMyClass(line: String) = {
    val split = line.split(',')
    transactions(split(0),split(1),split(2),split(3))
}

val  csv = sc.textFile("FILE").map(csvToMyClass)
csv.take(10)
csv.saveAsTextFile("PATH/output.csv")

How can I obtain the list of products associated group by Transaction_ID??

$\endgroup$
4
  • $\begingroup$ Which version of spark are you using ? $\endgroup$ – eliasah Aug 31 '16 at 9:39
  • $\begingroup$ Spark version 1.6.0 $\endgroup$ – SaCvP Aug 31 '16 at 9:46
  • $\begingroup$ ok why don't you use dataframes ? and spark-csv to read your csv ? is there a constraint on using RDDs ? $\endgroup$ – eliasah Aug 31 '16 at 9:46
  • $\begingroup$ Is there any advantage on using Data Frames? I don't have any constraint on using RDDs but I'm getting a little confusing on my code. I only want to "group" all the products based on transaction_ID $\endgroup$ – SaCvP Aug 31 '16 at 9:54
0
$\begingroup$

Since you are using Spark 1.6, I'd rather do these kind of transformations with DataFrame as it's much easier to manipulate.

You'll need to use SQLContext implicits for this :

import sqlContext.implicits._ // not need in spark-shell

Now, let's create some dummy data just to follow the code snippet that you have provided :

scala> val data = sc.parallelize(Seq("1,2,3,4", "2,3,4,5", "1,3,4,5", "1,6,6,7"))
// data: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[13] at parallelize at <console>:27

Back to your code, we define a case class Transactions and the CSV converter :

scala> case class Transactions(Transaction_ID: String, Customer_ID: String, Department: String, Product_ID: String)
// defined class Transactions

scala> def csvToMyClass(line: String) = { val split = line.split(','); Transactions(split(0), split(1), split(2), split(3)) }
// csvToMyClass: (line: String)Transactions

We will use implicits now to convert our data into a DataFrame after converting it to Transactions :

scala> val df = data.map(csvToMyClass).toDF("Transaction_ID", "Customer_ID", "Department", "Product_ID")
// df: org.apache.spark.sql.DataFrame = [Transaction_ID: string, Customer_ID: string, Department: string, Product_ID: string]

Let's take a look at the DataFrame :

scala> df.show
// +--------------+-----------+----------+----------+
// |Transaction_ID|Customer_ID|Department|Product_ID|
// +--------------+-----------+----------+----------+
// |             1|          2|         3|         4|
// |             2|          3|         4|         5|
// |             1|          3|         4|         5|
// |             1|          6|         6|         7|
// +--------------+-----------+----------+----------+

Now all we have to do is a simple group by and perform a collect_list aggregation on the first dataframe :

scala> val df2 = df.groupBy("Transaction_ID").agg(collect_list($"Product_ID"))
// df2: org.apache.spark.sql.DataFrame = [Transaction_ID: string, collect_list(Product_ID): array<string>]

We can check the content of our new DataFrame df2 :

scala> df.groupBy("Transaction_ID").agg(collect_list($"Product_ID")).show
// +--------------+------------------------+
// |Transaction_ID|collect_list(Product_ID)|
// +--------------+------------------------+
// |             1|               [4, 5, 7]|
// |             2|                     [5]|
// +--------------+------------------------+

I hope that this answers your question.

Note: If you wish to know what's the difference between RDD and DataFrames, I advice you to read Databrick's blog entry about it here.

$\endgroup$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.