【问题标题】：Filtering data from RDD using value sequence spark使用值序列火花从 RDD 中过滤数据
【发布时间】：2021-04-04 22:32:25
【问题描述】：

我需要以下用例的帮助：

问题 1： 我的 RDD 格式如下。现在从这个 RDD 中，我想排除 airport.code in("PUN","HAR","KAS")

    case class airport(code:String,city:String,airportname:String)
    val airportRdd=sparkSession.sparkContext.textFile("src/main/resources/airport_data.csv").
          map(x=>x.split(","))
    val airPortRddTransformed=airportRdd.map(x=>airport(x(0),x(1),x(2)))
    val trasnformedRdd=airPortRddTransformed.filter(air=>!(air.code.contains(seqValues:_*)))

但是！不工作。它告诉无法解析符号！。有人可以帮助我。如何在 RDD 中做否定。我只能使用 RDD 方法。

还有一个问题：

问题 2： 数据文件有 70 列。我有一个列序列：

val seqColumns=List("lat","longi","height","country")

我想在加载 RDD 时排除这些列。我该怎么做。我的生产 RDD 有 70 列，我真的知道要排除的列名。不是每列的索引。再次在其中查找RDD 方法。我知道如何在 Dataframe 方法中做到这一点。

【问题讨论】：

如果您分别提出每个问题会更好。也尝试提供一些airport_data.csv的内容或架构示例

标签： scala apache-spark apache-spark-sql rdd

【解决方案1】：

问题 1

使用broadcast 将值列表传递给filter 函数。过滤器中的 _* 似乎不起作用。我将条件更改为!seqValues.value.contains(air.code)

数据样本：airport_data.csv

C001,Pune,Pune Airport
C002,Mumbai,Chhatrapati Shivaji Maharaj International Airport
C003,New York,New York Airport
C004,Delhi,Delhi Airport

代码 sn-p

case class airport(code:String,city:String,airportname:String)
val seqValues=spark.sparkContext.broadcast(List("C001","C003"))
val airportRdd = spark.sparkContext.textFile("D:\\DataAnalysis\\airport_data.csv").map(x=>x.split(","))
val airPortRddTransformed = airportRdd.map(x=>airport(x(0),x(1),x(2)))
//airPortRddTransformed.foreach(println)
val trasnformedRdd = airPortRddTransformed.filter(air => !seqValues.value.contains(air.code))
trasnformedRdd.foreach(println)

输出 ->

airport(C002,Mumbai,Chhatrapati Shivaji Maharaj International Airport)
airport(C004,Delhi,Delhi Airport)

【讨论】：

gr8..非常感谢..让我试试

【解决方案2】：

我会改变的事情：

1- 您正在将 .csv 作为 TextFile 读取，然后根据 , 拆分行。您可以通过读取以下文件来保存此步骤：

val df = spark.read.csv("src/main/resources/airport_data.csv")

2- 更改contains的顺序

val trasnformedRdd = airPortRddTransformed.filter(air => !(seqValues.contains(air.code)))

【讨论】：