星火使用reduceByKey

Question

我的数据是这样的：

customer1表| ITEM1：X1，X2，X3; ITEM2：X1，X4，X5; item1：x1，x3，x6 | time1 | url
Customer1 | ITEM1：X1，X7，X3; ITEM2：X1，X4，X5; item3：x5 | time2 | url2
Customer2 | ITEM1：X1，X7，X3;项目3：X5 |时间3 | URL3

我想ReduceByKey相同customerIds和mapValues获得不同项目的工会为每个客户ID：

customer1表| ITEM1：X1，X2，X3; ITEM2：X1，X4，X5; ITEM1：X1，X3，X6; ITEM1：X1，X7，X3;项目3：X5

此我能够通过实现：

VAL线= spark.sparkContext.textFile（参数（0））
VAL记录= line.map（L =>升（x，y）=> x.union（y））。mapValues（x => .split（“\ |”））。map（l =>（l（0），l（1）））。reduceByKey x.distinct）

现在，我想在第二列中每个项目是唯一的，以及在同一个键的所有值应使用工会和独特，拿东西等加入：

Customer1 | ITEM1：X1，X2，X3，X6，X7; ITEM2：X1，X4，X5;项目3：X5

一旦做到这一点我想选择每个x的所有频率，例如：X1：2，X 2：1 .... 和更新的X（1-10）的向量对CustomerID与我得到的频率。

这可以在火花中实现吗？

Answer 1

是的，你当然可以在Spark中做到这一点！然而，你接近这个问题的方式使得它看起来有点难以实现。

所以我可以显示一个完整的复制pastable到REPL例子，让我们假设你的数据存储在一个字符串（不ARGS（0）文件）

val data = """Customer1| item1:x1,x2,x3; item2:x1,x4,x5; item1:x1,x3,x6|time1|url 
Customer1| item1:x1,x7,x3; item2:x1,x4,x5; item3:x5|time2|url2 
Customer2| item1:x1,x7,x3; item3:x5|time3|url3""" 

和RDD你所说的“线”，可以被读入RDD“rdd”为

val rdd = sc.parallelize(data.split("\n")) 

到目前为止没有新的东西。下一步是重要的一步。我们可以准备我们的数据一次完成所有工作，而不是分层计数和汇总。这是更可读性，也更有效，因为它是一个单一的地图，然后是一个单一的减少。

val mapped= rdd.flatMap(line => { 
    val arr = line.split("\\|") 
    val customer = arr(0) 
    val items = arr(1) 
    val time = arr(2) 
    val url = arr(3) 

    items.split(";").flatMap(item => { 
     val itemKey = item.split(":")(0) 
     val itemValues = item.split(":")(1).split(",") 

     itemValues.map(value => (customer, itemKey, value, time, url)) 
    }) 
}) 

我们可以看到什么是在这一点，我们可以用mapped.toDF("customer", "itemId", "itemValue", "time", "url").show

+---------+------+---------+-----+----+ 
| customer|itemId|itemValue| time| url| 
+---------+------+---------+-----+----+ 
|Customer1| item1|  x1|time1| url| 
|Customer1| item1|  x2|time1| url| 
|Customer1| item1|  x3|time1| url| 
|Customer1| item2|  x1|time1| url| 
|Customer1| item2|  x4|time1| url| 
|Customer1| item2|  x5|time1| url| 
|Customer1| item1|  x1|time1| url| 
|Customer1| item1|  x3|time1| url| 
|Customer1| item1|  x6|time1| url| 
|Customer1| item1|  x1|time2|url2| 
|Customer1| item1|  x7|time2|url2| 
|Customer1| item1|  x3|time2|url2| 
|Customer1| item2|  x1|time2|url2| 
|Customer1| item2|  x4|time2|url2| 
|Customer1| item2|  x5|time2|url2| 
|Customer1| item3|  x5|time2|url2| 
|Customer2| item1|  x1|time3|url3| 
|Customer2| item1|  x7|time3|url3| 
|Customer2| item1|  x3|time3|url3| 
|Customer2| item3|  x5|time3|url3| 
+---------+------+---------+-----+----+ 

很好地打印出来。最后，我们可以指望，减少到你所需要的载体：

val reduced = mapped.map{case (customer, itemKey, itemValue, time, url) => ((customer, itemKey, itemValue), 1)}. 
    reduceByKey(_+_). 
    map{case ((customer, itemKey, itemValue), count) => (customer, itemKey, itemValue, count)} 

并查看它：reduced.toDF("customer", "itemKey", "itemValue", "count").show

+---------+-------+---------+-----+            
| customer|itemKey|itemValue|count| 
+---------+-------+---------+-----+ 
|Customer1| item1|  x2| 1| 
|Customer1| item1|  x1| 3| 
|Customer2| item1|  x7| 1| 
|Customer1| item1|  x6| 1| 
|Customer1| item1|  x7| 1| 
|Customer2| item1|  x3| 1| 
|Customer2| item3|  x5| 1| 
|Customer1| item2|  x5| 2| 
|Customer1| item2|  x4| 2| 
|Customer1| item2|  x1| 2| 
|Customer1| item3|  x5| 1| 
|Customer1| item1|  x3| 3| 
|Customer2| item1|  x1| 1| 
+---------+-------+---------+-----+ 

如果您需要将所有组合到矢量的Array/Seq表示中，则可以通过进一步聚合数据来完成此操作。希望这可以帮助！