如何使用scala过滤火花壳中的数据？

Question

我有以下数据，需要使用星星之火(Scala)进行排序，这样，我只需要访问“沃尔玛”的人的身份，而不需要“百思买”。存储可能是重复的，因为一个人可以访问商店的任何次数。

输入数据：

身份证，商店

1，沃尔玛

1，沃尔玛

1，百思买

2，目标

3，沃尔玛

4，百思买

预期产出: 3，沃尔玛

我已经使用dataFrames获得了输出，并在spark上下文上运行了SQL查询。但是，在没有groupByKey/reduceByKey的情况下，是否可以使用dataFrames /reduceByKey等来实现这一目的呢？有人能帮我处理代码吗?在map-> groupByKey之后，ShuffleRDD已经形成，我在过滤CompactBuffer时遇到了困难！

使用sqlContext 获得的代码如下所示：

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext.createSchemaRDD

case class Person(id: Int, store: String)

val people = sc.textFile("examples/src/main/resources/people.txt")
               .map(_.split(","))
               .map(p => Person(p(1)trim.toInt, p(1)))
people.registerTempTable("people")

val result = sqlContext.sql("select id, store from people left semi join (select id from people where store in('Walmart','Bestbuy') group by id having count(distinct store)=1) sample on people.id=sample.id and people.url='Walmart'")

我现在正在尝试的代码是这样的，但是在第三步之后，我被击中了：

val data = sc.textFile("examples/src/main/resources/people.txt")
             .map(x=> (x.split(",")(0),x.split(",")(1)))
             .filter(!_.filter("id"))
val dataGroup = data.groupByKey()
val dataFiltered = dataGroup.map{case (x,y) =>
    val url = y.flatMap(x=> x.split(",")).toList
    if (!url.contains("Bestbuy") && url.contains("Walmart")){
        x.map(x=> (x,y))}}

如果我执行dataFiltered.collect()，我得到的是ArrayAny =Array(向量(3，沃尔玛))，()，()

请帮助我在这个步骤之后提取输出。

Answer 1

若要筛选RDD，只需使用RDD.filter

val dataGroup = data.groupByKey()

val dataFiltered = dataGroup.filter {
  // keep only lists that contain Walmart but do not contain Bestbuy:
  case (x, y) => val l = y.toList; l.contains("Walmart") && !l.contains("Bestbuy")
}

dataFiltered.foreach(println) // prints: (3,CompactBuffer(Walmart))

// if you want to flatten this back to tuples of (id, store):
val result = dataFiltered.flatMap { case (id, stores) => stores.map(store => (id, store)) }

result.foreach(println) // prints: (3, Walmart)

Answer 2

我也尝试了另一种方法，结果成功了。

val data = sc.textFile("examples/src/main/resources/people.txt")
     .filter(!_.filter("id"))
         .map(x=> (x.split(",")(0),x.split(",")(1)))
data.cache()
val dataWalmart = data.filter{case (x,y) => y.contains("Walmart")}.distinct()
val dataBestbuy = data.filter{case (x,y) => y.contains("Bestbuy")}.distinct()
val result = dataWalmart.subtractByKey(dataBestbuy)
data.uncache()