调用 PySpark 数据框列表中的项目

Question

我有以下数据框

+------------------+----------+------------------+
|        antecedent|consequent|        confidence|
+------------------+----------+------------------+
|         [7, 2, 0]|       [8]|0.6237623762376238|
|         [7, 2, 0]|       [1]|               1.0|
|         [7, 2, 0]|       [5]|0.9975247524752475|
|         [7, 2, 0]|       [3]|0.9975247524752475|
|         [7, 2, 0]|       [4]|0.9975247524752475|
|         [7, 2, 0]|       [6]| 0.995049504950495|
|      [6, 5, 3, 4]|       [8]| 0.623721881390593|
|      [6, 5, 3, 4]|       [1]|               1.0|
|      [6, 5, 3, 4]|       [2]|               1.0|
|      [6, 5, 3, 4]|       [0]|               1.0|
|      [6, 5, 3, 4]|       [7]| 0.820040899795501|
|[9, 8, 6, 5, 1, 2]|       [0]|               1.0|
|[9, 8, 6, 5, 1, 2]|       [3]|               1.0|
|[9, 8, 6, 5, 1, 2]|       [4]|               1.0|
|         [7, 3, 1]|       [8]|0.6228287841191067|
|         [7, 3, 1]|       [5]|               1.0|
|         [7, 3, 1]|       [2]|               1.0|
|         [7, 3, 1]|       [0]|               1.0|
|         [7, 3, 1]|       [4]|               1.0|
|         [7, 3, 1]|       [6]|0.9950372208436724|
+------------------+----------+------------------+

我想对其进行一些查询，例如过滤前行不包含 [7,3] 的行，我尝试了此查询，但似乎错误，因为 7,3 是整数

from pyspark.sql.functions import *
q = r.filter(~col('antecedent').isin([7,3])).show()

错误：

"condition should be string or Column"

Answer 1

您可以编写一个 udf 函数来检查条件（检查 7 或 3 是否存在于先行词 column 中的条件）为

from pyspark.sql import functions as F
from pyspark.sql import types as T
def checkIsIn(array):
    return True in [x in array for x in [7, 3]]

udfCheckIsIn = F.udf(checkIsIn, T.BooleanType())

然后在过滤器中使用它作为

q = r.filter(udfCheckIsIn(r.antecedent)).show()

你的输出应该是

+------------+----------+------------------+
|  antecedent|consequent|        confidence|
+------------+----------+------------------+
|   [7, 2, 0]|       [8]|0.6237623762376238|
|   [7, 2, 0]|       [1]|               1.0|
|   [7, 2, 0]|       [5]|0.9975247524752475|
|   [7, 2, 0]|       [3]|0.9975247524752475|
|   [7, 2, 0]|       [4]|0.9975247524752475|
|   [7, 2, 0]|       [6]| 0.995049504950495|
|[6, 5, 3, 4]|       [8]| 0.623721881390593|
|[6, 5, 3, 4]|       [1]|               1.0|
|[6, 5, 3, 4]|       [2]|               1.0|
|[6, 5, 3, 4]|       [0]|               1.0|
|[6, 5, 3, 4]|       [7]| 0.820040899795501|
|   [7, 3, 1]|       [8]|0.6228287841191067|
|   [7, 3, 1]|       [5]|               1.0|
|   [7, 3, 1]|       [2]|               1.0|
|   [7, 3, 1]|       [0]|               1.0|
|   [7, 3, 1]|       [4]|               1.0|
|   [7, 3, 1]|       [6]|0.9950372208436724|
+------------+----------+------------------+

用于检查条件（检查 7 和 3 是否存在于先行词 column 中的条件）为

from pyspark.sql import functions as F
from pyspark.sql import types as T
def checkIsIn(array):
    return False not in [x in array for x in [7, 3]]

udfCheckIsIn = F.udf(checkIsIn, T.BooleanType())

然后在过滤器中使用它作为

q = r.filter(udfCheckIsIn(r.antecedent)).show()

你的输出应该是

+----------+----------+------------------+
|antecedent|consequent|        confidence|
+----------+----------+------------------+
| [7, 3, 1]|       [8]|0.6228287841191067|
| [7, 3, 1]|       [5]|               1.0|
| [7, 3, 1]|       [2]|               1.0|
| [7, 3, 1]|       [0]|               1.0|
| [7, 3, 1]|       [4]|               1.0|
| [7, 3, 1]|       [6]|0.9950372208436724|
+----------+----------+------------------+