python - 从一个 PySpark 数据帧中获取 ArrayType 列，并在另一个数据帧中获取相应的值

Question

我有两个数据框，一个称为项目集，另一个称为运行 FPGrowth 的规则。它们的格式如下:

ITEMSETS DATAFRAME
+------------------------+-----+
|cart                    |freq |
+------------------------+-----+
|[7084781116]            |10492|
|[7084781116, 2606500532]|362  |
|[7084781116, 0]         |327  |
|[7084781116, 2001]      |393  |
|[7084781116, 4011]      |489  |
|[4460030705]            |448  |
|[3800035800]            |539  |
|[3022304060]            |1188 |
|[2800021730]            |901  |
|[1380018805]            |437  |
+------------------------+-----+

RULES DATAFRAME
+--------------------+------------+
|          antecedent|  consequent|
+--------------------+------------+
|        [1686, 4068]|[2640000010]|
|[1675, 4432, 3680...|      [1673]|
|[1676, 1677, 3680...|      [1678]|
|[1676, 1677, 3680...|[3680080816]|
|[1676, 1677, 3680...|[3680044476]|
|[1676, 1677, 3680...|      [1675]|
|[7283040006, 7110...|      [1683]|
|[7283040006, 7110...|      [1682]|
|[1674, 4300000953...|      [1673]|
|[1674, 4300000953...|[3680044476]|
+--------------------+------------+

我需要添加一些新列来执行计算(如果您好奇的话，可以颠倒前因和后果关系。)

首先，我需要向规则数据框中添加一个新列，该列具有项集数据框中的相应频率。例如，如果结果是 [7084781116]，新列将具有项集数据帧中该数组的频率(10492，通过第一行。)

接下来，我需要将后件的值附加到先件，然后执行相同的操作。例如，查看规则列中的第二行，我需要将 1673 添加到 [1675, 4432, 3680...]，然后从项集表中获取该数组的频率，并将其存储在另一列中。

有人可以帮我解决这个问题吗？我对 PySpark 还很陌生，无法理解。我尝试实现几个 UDF，例如使用以下内容，希望将数组转换为字符串，以便可能使它们更易于使用:

In [6]: conv_to_str([1,3,2,6,5])
Out[6]: '1|2|3|5|6|'

In [7]: def conv_to_str(listname):
        listname.sort()
        rv = ""
        for val in listname:
            rv += str(val) + "|"
        return rv[:-1]

In [8]: conv_to_str([1,3,2,6,5])
Out[8]: '1|2|3|5|6' 

谢谢!

Answer 1

我建议采取以下三个步骤:

1. 要添加频率列，请在 consequent 和 cart 列上使用左连接。
2. 要将consequent值添加到antecendent数组，请使用concat函数(从 Spark 2.4 开始支持数组)。
3. 在丰富的antecendent 列上再次执行左连接，以获取此连接数组的频率。

因此，在 PySpark 中，查询可能如下所示:

(
  rules_df
  .withColumn('antecedent_enriched', concat('antecendent', 'consequent'[0]))
  .alias('a')
  .join(itemsets_df.alias('b'), col('a.consequent') == col('b.cart'), 'left')
  .join(itemsets_df.alias('c'), col('a.antecedent_enriched') == col('c.cart'), 'left'))
  .select(
    'antecedent',
    'consequent',
    'b.freq',
    'antecedent_enriched',
    col('c.freq').alias('freq_enriched')
  )
)

使用 concat 函数时也要小心，因为如果 consequent 列包含 Null 值，则串联结果也将为 Null。