问题描述
假设您具有以下pyspark DataFrame:
Suppose you had the following pyspark DataFrame:
data= [('foo',), ('123',), (None,), ('bar',)]
df = sqlCtx.createDataFrame(data, ["col"])
df.show()
#+----+
#| col|
#+----+
#| foo|
#| 123|
#|null|
#| bar|
#+----+
接下来的两个代码块应该做同样的事情-也就是说，如果不是null，则返回该列的大写字母.但是，第二种方法(使用udf)会产生错误.
The next two code blocks should do the same thing- that is, return the uppercase of the column if it is not null. However, the second method (using a udf) produces an error.
方法1 :使用pyspark.sql.functions.upper()
import pyspark.sql.functions as f
df.withColumn(
'upper',
f.when(
f.isnull(f.col('col')),
f.col('col')
).otherwise(f.upper(f.col('col')))
).show()
#+----+-----+
#| col|upper|
#+----+-----+
#| foo| FOO|
#| 123| 123|
#|null| null|
#| bar| BAR|
#+----+-----+
方法2 :在udf
df.withColumn(
'upper',
f.when(
f.isnull(f.col('col')),
f.col('col')
).otherwise(f.udf(lambda x: x.upper(), StringType())(f.col('col')))
).show()
这给了我AttributeError: 'NoneType' object has no attribute 'upper'.为什么对when的调用中的f.isnull()检查似乎被忽略了?
This gives me AttributeError: 'NoneType' object has no attribute 'upper'. Why is the f.isnull() check in the call to when seemingly being ignored?
我知道我可以将udf更改为f.udf(lambda x: x.upper() if x else x, StringType())来避免此错误，但是我想了解为什么会发生这种情况.
I know that I can change my udf to f.udf(lambda x: x.upper() if x else x, StringType()) to avoid this error, but I'd like to understand why it's happening.
完整追溯:
Py4JJavaErrorTraceback (most recent call last)
<ipython-input-38-cbf0ffe73538> in <module>()
4 f.isnull(f.col('col')),
5 f.col('col')
----> 6 ).otherwise(f.udf(lambda x: x.upper(), StringType())(f.col('col')))
7 ).show()
/opt/SPARK2/lib/spark2/python/pyspark/sql/dataframe.py in show(self, n, truncate)
316 """
317 if isinstance(truncate, bool) and truncate:
--> 318 print(self._jdf.showString(n, 20))
319 else:
320 print(self._jdf.showString(n, int(truncate)))
/opt/SPARK2/lib/spark2/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py in __call__(self, *args)
1131 answer = self.gateway_client.send_command(command)
1132 return_value = get_return_value(
-> 1133 answer, self.gateway_client, self.target_id, self.name)
1134
1135 for temp_arg in temp_args:
/opt/SPARK2/lib/spark2/python/pyspark/sql/utils.py in deco(*a, **kw)
61 def deco(*a, **kw):
62 try:
---> 63 return f(*a, **kw)
64 except py4j.protocol.Py4JJavaError as e:
65 s = e.java_exception.toString()
/opt/SPARK2/lib/spark2/python/lib/py4j-0.10.4-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
317 raise Py4JJavaError(
318 "An error occurred while calling {0}{1}{2}.\n".
--> 319 format(target_id, ".", name), value)
320 else:
321 raise Py4JError(
Py4JJavaError: An error occurred while calling o642.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 51 in stage 77.0 failed 4 times, most recent failure: Lost task 51.3 in stage 77.0 (TID 5101, someserver.prod.somecompany.net, executor 99): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/opt/SPARK2/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py", line 174, in main
process()
File "/opt/SPARK2/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py", line 169, in process
serializer.dump_stream(func(split_index, iterator), outfile)
File "/opt/SPARK2/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py", line 106, in <lambda>
func = lambda _, it: map(mapper, it)
File "/opt/SPARK2/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py", line 92, in <lambda>
mapper = lambda a: udf(*a)
File "/opt/SPARK2/lib/spark2/python/lib/pyspark.zip/pyspark/worker.py", line 70, in <lambda>
return lambda *a: f(*a)
File "<ipython-input-38-cbf0ffe73538>", line 6, in <lambda>
AttributeError: 'NoneType' object has no attribute 'upper'
推荐答案
您必须记住，Spark SQL(与RDD不同)并不是您所看到的.优化器/计划器可以按任意顺序自由安排操作，甚至可以重复执行多个阶段.
You have to remember that Spark SQL (unlike RDD) is not what-you-see-is-what-you-get. Optimizer / planner is free to schedule operations in the arbitrary order or even repeat stages multiple times.
Python udfs并非基于Row应用，而是使用批处理模式. when并没有太多被忽略，但并未用于优化执行计划:
Python udfs are not applied on a Row basis, but using batch mode. when is not so much ignored, but not used to optimize execution plan:
== Physical Plan ==
*Project [col#0, CASE WHEN isnull(col#0) THEN col#0 ELSE pythonUDF0#21 END AS upper#17]
+- BatchEvalPython [<lambda>(col#0)], [col#0, pythonUDF0#21]
+- Scan ExistingRDD[col#0]
因此，与udf一起使用的功能必须对None输入具有鲁棒性，例如:
Therefore function used with udf has to be robust to None inputs, for example:
df.withColumn(
'upper',
f.udf(
lambda x: x.upper() if x is not None else None,
StringType()
)(f.col('col'))
).show()
这篇关于使用UDF忽略条件何时的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！