Spark-r manupulating数据

Question

我有一个如下样本数据：

UserId,ProductId,Category,Action
1,111,Electronics,Browse
2,112,Fashion,Click
3,113,Kids,AddtoCart
4,114,Food,Purchase
5,115,Books,Logout
6,114,Food,Click
7,113,Kids,AddtoCart
8,115,Books,Purchase
9,111,Electronics,Click
10,112,Fashion,Purchase
3,112,Fashion,Click

我需要生成对“时尚”类别或“电子”类别感兴趣但不在两个类别中都感兴趣的用户列表。如果他/她使用我已完成的火花/标量代码执行了任何这些操作（Click / AddToCart / Purchase），用户会感兴趣：

val rrd1 = sc.textFile("/user/harshit.kacker/datametica_logs.csv")
val rrd2 = rrd1.map( x=> {
     | val c = x.split(",")
     | (c(0).toInt , x)})

val rrd3 = rrd1.filter(x=> x.split(",")(2) == "Fashion" || x.split(",")(2) == "Electronics")
val rrd4 = rrd3.filter(x=> x.split(",")(3)== "Click" || x.split(",")(3)=="Purchase" || x.split(",")(3)=="AddtoCart")
rrd4.collect.foreach(println)

2,112,Fashion,Click
9,111,Electronics,Click
10,112,Fashion,Purchase
3,112,Fashion,Click
4,111,Electronics,Click
19,112,Fashion,Click
9,112,Fashion,Purchase
2,112,Fashion,Click
2,111,Electronics,Click
1,112,Fashion,Purchase

现在我必须努力“生成对”时尚“类别或”电子“类别感兴趣但不在两个类别中的用户列表”这个斜体部分并得到所需的输出：

10,Fashion
3,Fashion
4,Electronics
19,Fashion
1,Fashion

意味着用户将时装和电子作为类别应该被淘汰，请你帮忙实现同样的目标？

Answer 1

首先将输入文本文件解析为元组：

val srcPath = "/user/harshit.kacker/datametica_logs.csv"

// parse test file in to tuples:
val rdd = spark.sparkContext.textFile(srcPath)
val rows = rdd.map(line => line.split(",")).map(row => (row(0), row(1), row(2), row(3)))
val header = rows.first
// drop the header:
val logs = rows.filter(row => row != header)

按兴趣标准过滤RDD：

val interests = logs.filter(log =>
  List("Click", "AddtoCart", "Purchase").contains(log._4)
)

分别过滤时装和电子产品：

val fashion = interests.filter(row => row._3 == "Fashion")
val electronics = interests.filter(row => row._3 == "Electronics")

找到时尚和电子产品之间的常见用户ID：

val fashionIds = fashion.map(_._1).distinct
val electronicsIds = electronics.map(_._1).distinct
val commonIds = fashionIds.intersection(electronicsIds).collect()

结合时尚和电子行并过滤两者之间共同的ID：

val finalRdd = (fashion ++ electronics)
  .filter(log => !commonIds.contains(log._1))
  .map(log => (log._1, log._3))
  .distinct()

编辑：使用DataFrame

// using dataframes:

val df = spark.read.option("header", "true").csv(srcPath)
val interestDf = df.where($"Action".isin("Click", "Purchase", "AddToCart"))
val fashionDf = interestDf.where($"Category" === "Fashion")
val electronicsDf = interestDf.where($"Category" === "Electronics")

val joinDf = electronicsDf.alias("e").join(fashionDf.alias("f"), Seq("UserId"), "outer")
  .where($"e.Category".isNull || $"f.Category".isNull)

val finalDf = joinDf.select($"UserId", when($"e.Category".isNull, $"f.Category").otherwise($"e.Category").as("Category")).distinct