1
我需要使用我自己的用户定义函数来过滤Spark数据帧。我的数据框是使用jdbc连接从数据库中读取的,然后在进行过滤之前通过spark中的自联接操作。在尝试过滤后的数据帧collect
时发生错误。自加入后使用UDF的Spark 2.0过滤器
我一直在Spark 1.6中成功使用它。然而,在升级到2.0后,昨日它失败,错误:
py4j.protocol.Py4JJavaError: An error occurred while calling o400.collectToPython.
: java.lang.UnsupportedOperationException: Cannot evaluate expression:
<lambda>(input[0, string, true])
这里是产生错误(在我的环境),一个小例子:
from pyspark.sql.functions import udf, col
from pyspark.sql.types import BooleanType
spark = SparkSession.builder.master('local').appName('test').getOrCreate()
# this works successfully
df = spark.createDataFrame([('Alice', 1), ('Bob', 2), ('Dan', None)],
['name', 'age'])
df.filter(udf(lambda x: 'i' in x, BooleanType())(df.name)).collect()
>>> [Row(name=u'Alice', age=1)]
# this produces the error
df_emp = spark.createDataFrame([(1, 'Alice', None), (2, 'Bob', 1),
(3, 'Dan', 2), (4, 'Joe', 2)],
['id', 'name', 'manager_id'])
df1 = df_emp.alias('df1')
df2 = df_emp.alias('df2')
cols = df1.columns
# the self-join
result = df1.join(df2, col('df1.id') == col('df2.manager_id'), 'left_outer')
result.collect()
>>> [Row(id=1, name=u'Alice', manager_id=None),
Row(id=3, name=u'Dan', manager_id=2), Row(id=2, name=u'Bob', manager_id=1),
Row(id=2, name=u'Bob', manager_id=1), Row(id=4, name=u'Joe', manager_id=2)]
# simple udf filter
filtered = result.filter(udf(lambda x: 'i' in x, BooleanType())(result.name))
filtered.collect()
# the above error is produced...
难道我做错什么在这种情况下, ?这是2.0中的一个错误还是应该考虑两个版本之间的行为改变?
我一直在敲我的头挂在墙上,因为在之前的会议上工作了UDF是失败。这救了我!谢谢蒂姆! –