Apply GroupByKey to identify the Frequently Products Purchase Together

Question

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

Answer 1

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.