使用 PySpark 按列名及其数值过滤 spark RDD答案

【问题标题】：Filter spark RDD with PySpark by column name and its numerical value使用 PySpark 按列名及其数值过滤 spark RDD
【发布时间】：2018-05-26 13:39:55
【问题描述】：

我正在将 Scala / Spark 模型翻译成 Python / Spark。问题是我有大约 100 万个观察值和大约 33 列的 RDD。我正在根据数值阈值（“时间”）拆分 RDD。时间变量是数字格式（双精度）（不是 posix）。

这里是 Scala 源代码：

// get the time to split the data.
val splitTime = data.stat.approxQuantile("Time", Array(0.7), 0.001).head

val trainingData = data.filter(s"Time<$splitTime").cache()
val validData = data.filter(s"Time>=$splitTime").cache()

这是我的 PySpark 失败的解释：

splitTime = data.approxQuantile("Time", [0.7], 0.001)
trainingData = data.filter(data["Time"] < splitTime)
validData = data.filter(data["Time"] >= splitTime)

第一行工作正常。当我尝试在 RDD 上使用阈值时，问题就开始了。我也无法解码围绕条件的 Scala 格式 s" >=$ " 及其在条件中的重要性。网上来源对s" >=$ "的含义很模糊。

【问题讨论】：

标签： python scala apache-spark pyspark rdd

【解决方案1】：

approxQuantile 返回List[float]（如这里的单列大小写）或List[List[float]]（多列大小写），因此您必须提取值：

splitTime = data.approxQuantile("Time", [0.7], 0.001)
data.filter(data["Time"] < splitTime[0])

或

(litTime, ) = data.approxQuantile("a", [0.7], 0.001)
trainingData = data.filter(data["Time"] < splitTime)

【讨论】：