pyspark 数据框过滤并没有真正删除行？答案

【问题标题】：pyspark dataframe filtering doesn't really remove rows?pyspark 数据框过滤并没有真正删除行？
【发布时间】：2018-07-06 11:10:07
【问题描述】：

我的数据框经过两次连续过滤，每次都使用布尔值 UDF。第一个过滤删除所有列不作为键出现在某些广播字典中的行。第二个过滤对该字典与当前键关联的值施加阈值。

如果我仅在第一次过滤后显示结果，则带有“c”的行不在其中，正如预期的那样。但是，尝试显示第二次过滤的结果会导致 u'c' 出现 KeyError 异常

sc = SparkContext()
ss = SparkSession(sc)

mydict={ "a" : 4, "b" : 6 }
mydict_bc = sc.broadcast(mydict)

udf_indict=func.udf( lambda x: x in mydict_bc.value, BooleanType() )
udf_bigenough=func.udf( lambda x: mydict_bc.value[x] > 5, BooleanType() )

df=ss.createDataFrame([ "a", "b", "c" ], StringType() ).toDF("name")
df1 = df.where( udf_indict('name') )
df1.show()

    +----+                                                                          
    |name|
    +----+
    |   a|
    |   b|
    +----+

df1.where( udf_bigenough('name') ).show()

KeyError: u'c'
at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193)
at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:234)
at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152)
at org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$doExecute$1.apply(BatchEvalPythonExec.scala:144)
at org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$doExecute$1.apply(BatchEvalPythonExec.scala:87)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:797)
    ...

我猜这与延迟执行和内部优化有关，但这真的是预期的行为吗？

谢谢

【问题讨论】：

标签： apache-spark dataframe pyspark apache-spark-sql user-defined-functions

【解决方案1】：

这个

我的数据框经过两次连续过滤

是不正确的假设。与RDD 不同，所有转换都是所见即所得，SQL API 是纯声明性的。它解释了必须做什么，但没有说明如何做。优化器可以根据需要重新排列所有元素。

使用非确定性变体将禁用优化：

df1 = df.where( udf_indict.asNondeterministic()('name'))
df1.where( udf_bigenough.asNondeterministic()('name') ).show()

但你真的应该处理异常

@udf(BooleanType())
   def udf_bigenough(x):
      try:
          return mydict_bc.get(x) > 5
      except TypeError:
          pass

或者更好，不要使用udf。

【讨论】：