spark-rdd操作数据

8ehkhllq 于 2021-05-29 发布在 Hadoop

关注(0)|答案(1)|浏览(369)

我有如下示例数据：

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

我需要生成的用户谁是感兴趣的“时尚”或“电子”类别，但不是在这两个类别列表。用户是否有兴趣使用spark/scala代码执行这些操作（单击/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

意味着时尚和电子类的用户ID应该被淘汰。我怎样才能达到同样的效果？

hadoop Hive scala apache-spark apache-spark-sql

来源：https://stackoverflow.com/questions/52192684/spark-rdd-manipulating-data

1条答案

按热度按时间

yqhsw0fo1#

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

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")

查找fashion和electronics之间的通用用户ID：

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

合并fashion和electronics行并过滤两者之间的共同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

赞(0）回复(0）举报 2021-05-29

我来回答

spark-rdd操作数据

1条答案

相关问题

热门标签

最新问答