使用pyspark在map阶段过滤rdd的行答案

【问题标题】：FIltering rows of an rdd in map phase using pyspark使用pyspark在map阶段过滤rdd的行
【发布时间】：2021-02-19 22:59:34
【问题描述】：

我在 pyspark 中使用此代码过滤数据集：

rdd = sc.textFile("location...").map(lambda line: line.split(",")). \
         filter(lambda line :condition...)

我的问题是：在我的解决方案伪代码中，可以在映射阶段完成对不符合我的条件的行的过滤，从而解析整个数据集一次。但是在这种情况下，数据集被解析2倍更贵。

有没有办法通过一次解析来做到这一点？

【问题讨论】：

数据集没有被解析两次，map的输出作为filter的输入。

标签： python apache-spark pyspark rdd

【解决方案1】：

由于您的代码映射部分是在过滤之前完成的，如果您想提供更多的优化并且您的映射函数输出不需要过滤，在这种情况下，建议在映射之前进行过滤，这样可以减少映射函数的输入元素的数量

映射前过滤

rdd = sc.textFile("location...").filter(lambda line: line.split(",")). \
     map(lambda line :condition...)

另外，如果您想在映射函数中提供一些过滤逻辑，可以这样做，但是在这种情况下，您需要在最后过滤 NONE 类型的元素。

words = sc.parallelize(
["my code",
 "java is a required",
 "hadoop is a framework",
 "spark",
 "akka",
 "spark vs hadoop",
 "pyspark",
 "pyspark and spark"]
)

def mapCondition(line):
   if(line.startswith("p")):
       return line

tokenized = words.map(lambda line: mapCondition(line))
print tokenized.collect()

【讨论】：

【解决方案2】：

您可以在过滤器lambda函数中直接对line.split(",")进行操作。例如，您可以比较第一个逗号之前的字符串，如下所示：

rdd = sc.textFile("location...").filter(lambda line: line.split(",")[0] = "string")

【讨论】：