PySpark 使用 spark 原生函数过滤 RDD答案

【问题标题】：PySpark filter RDD using spark native functionsPySpark 使用 spark 原生函数过滤 RDD
【发布时间】：2021-05-21 08:35:54
【问题描述】：

使用 pySpark，我需要过滤作为字符串列表的 RDD：

In [74]: sc.textFile("*.txt").collect()
Out[74]:
['laber\tblubber',
 'foo\tbar',
 'dummy\tdumdum',
 'col1\tcol2\tcol3\tcol4\tcol5',
 ' 1\t2\t3\t4\t5',
 ' 11\t22\t33\t44\t44',
 ' 9\t8\t7\t6\t5',
 'laber\tblubber',
 'foo\tbar',
 'dummy\tdumdum',
 'col1\tcol2\tcol3\tcol4\tcol5',
 ' 99\t2\t3\t4\t5',
 ' 99\t22\t33\t44\t44',
 ' 99\t8\t7\t6\t5']

我想过滤掉任何不以空格开头的行。我知道，我可以通过以下方式实现：

sc.textFile("*.txt").filter(lambda x: x[0] == " ")

但是我希望获得最高性能，据我了解，使用 python lamdas 会增加开销，并且查询计划器无法很好地优化。

如何在 RDD 上使用 spark 原生函数？

我期待这样的事情：

sc.textFile("*.txt").filter("substr(_, 0, 1) == ' '")

【问题讨论】：

Native Spark SQL 函数适用于数据帧。你可以使用数据帧而不是 RDD 吗？
我可以。往返 rdd -> df -> rdd 的性能影响如何（我需要一个 rdd 来反馈给 CSV 解析器，出于性能和边缘情况的原因，我不热衷于使用正则表达式等来实现自己）
不确定它是否能提供更好的性能，但你可以试试spark.createDataFrame(sc.textFile("*.txt").map(Row)).filter("substring(_1, 1, 1) != ' '").rdd.map(lambda x: x[0])
你能发布原始输入文件吗，通过查看你的输出，我相信我们应该能够在数据框中使用“spark.read.csv”和“\t”来读取这个文件分隔符，一旦我们在数据框中，我们应该能够使用原生 spark 过滤掉。

标签： apache-spark pyspark apache-spark-sql

【解决方案1】：

您可以使用 spark SQL 函数，例如：

df = spark.sql("""
SELECT line FROM text.`./`
WHERE line NOT LIKE ' %'
""")

我从未加载过这样的文本文件（主要是 parquet、JSON 或 CSV），但我相信它也应该可以工作。看看this spark SQL docs entry。

【讨论】：

问题是关于rdds
是的，还有针对性能优化的本机功能。我相信 SQL 函数是要走的路。
显然最好的办法是查看不同方法的基准
Rdds 是老帽子