Pyspark我们如何检查列表中是否包含列值

Question

我试图弄清楚是否有一个函数可以检查Spark DataFrame的一列是否包含列表中的任何值：

# define a dataframe
rdd = sc.parallelize([(0,100), (0,1), (0,2), (1,2), (1,10), (1,20), (3,18), (3,18), (3,18)])
df = sqlContext.createDataFrame(rdd, ["id", "score"])
# define a list of scores
l = [1]
# filter out records by scores by list l
records = df.filter(~df.score.contains(l))
# expected: (0,100), (0,1), (1,10), (3,18)

我遇到一个运行此代码的问题：

java.lang.RuntimeException: Unsupported literal type class java.util.ArrayList [1]

有没有办法做到这一点，还是我们必须循环遍历列表才能包含？

Answer 1

我在不使用udf的情况下看到了一些方法。

您可以使用pyspark.sql.functions.regexp_extract使用列表理解，并利用这样的事实，即如果没有匹配，将返回一个空字符串。

尝试提取列表中的所有值l中的所有值并将结果串联。如果所得的串联字符串是一个空字符串，则意味着不匹配的值。

例如：

from pyspark.sql.functions import concat, regexp_extract
records = df.where(concat(*[regexp_extract("score", str(val), 0) for val in l]) != "")
records.show()
#+---+-----+
#| id|score|
#+---+-----+
#|  0|  100|
#|  0|    1|
#|  1|   10|
#|  3|   18|
#|  3|   18|
#|  3|   18|
#+---+-----+

如果您查看执行计划，您会看到它足够聪明，将score列施放到string：

records.explain()
#== Physical Plan ==
#*Filter NOT (concat(regexp_extract(cast(score#11L as string), 1, 0)) = )
#+- Scan ExistingRDD[id#10L,score#11L]

另一种方法是使用pyspark.sql.Column.like（或类似地与rlike）：

from functools import reduce
from pyspark.sql.functions import col
records = df.where(
    reduce(
        lambda a, b: a|b, 
        map(
            lambda val: col("score").like(val.join(["%", "%"])), 
            map(str, l)
        )
    )
)

产生与上述相同的输出，并具有以下执行计划：

#== Physical Plan ==
#*Filter Contains(cast(score#11L as string), 1)
#+- Scan ExistingRDD[id#10L,score#11L]

---

如果您只想要不同的记录，则可以做：

records.distinct().show()
#+---+-----+
#| id|score|
#+---+-----+
#|  0|    1|
#|  0|  100|
#|  3|   18|
#|  1|   10|
#+---+-----+

Answer 2

如果我正确理解您，您想拥有一个列表，其中包含元素的唯一1。您想检查此元素是否出现在分数中。在这种情况下，它更容易使用字符串，而不是直接与数字一起使用。

您可以使用自定义地图功能来执行此操作，并通过UDF应用（直接申请导致某些奇怪的行为，有时仅能工作）。

找到以下代码：

rdd = sc.parallelize([(0,100), (0,1), (0,2), (1,2), (1,10), (1,20), (3,18), (3,18), (3,18)])
df = sqlContext.createDataFrame(rdd, ["id", "score"])
l = [1]
def filter_list(score, l):
    found = True
    for e in l:
        if str(e) not in str(score):  #The filter that checks if an Element e
            found = False             #does not appear in the score
    if found:
        return True                   #boolean value if the all elements were found
    else:
        return False
def udf_filter(l):
    return udf(lambda score: filter_list(score, l)) #make a udf function out of the filter list
df.withColumn("filtered", udf_filter(l)(col("score"))).filter(col("filtered")==True).drop("filtered").show()
#apply the function and store results in "filtered" column afterwards 
#only select the successful filtered rows and drop the column

输出：

+---+-----+
| id|score|
+---+-----+
|  0|  100|
|  0|    1|
|  1|   10|
|  3|   18|
|  3|   18|
|  3|   18|
+---+-----+