以下代码可完成所请求的任务。定义了一个用户自定义函数,该函数接收DataFrame
的两列作为参数。因此,对于每一行,搜索项是否在项列表中。如果找到该项,则返回1,否则返回0。
from pyspark.sql.types import IntegerType
from pyspark.sql.functions import udf
rdd = sc.parallelize([('u1', 1, [1 ,2, 3]), ('u1', 4, [1, 2, 3])])
df = rdd.toDF(['user', 'item', 'fav_items'])
df.show()
function = udf(lambda item, items: 1 if item in items else 0, IntegerType())
df.select('user', 'item', 'fav_items', function(col('item'), col('fav_items')).alias('result')).show()
以下是结果:
+
|user|item|fav_items|
+
| u1| 1|[1, 2, 3]|
| u1| 4|[1, 2, 3]|
+
+
|user|item|fav_items|result|
+
| u1| 1|[1, 2, 3]| 1|
| u1| 4|[1, 2, 3]| 0|
+