PySpark : Comparer les valeurs d'un tableau dans un dataFrame avec les valeurs d'un tableau dans un autre dataFrame pour obtenir l'intersection.

Question

J'ai les deux DataFrames suivants :

l1 = [(['hello','world'],), (['stack','overflow'],), (['hello', 'alice'],), (['sample', 'text'],)]
df1 = spark.createDataFrame(l1)

l2 = [(['big','world'],), (['sample','overflow', 'alice', 'text', 'bob'],), (['hello', 'sample'],)]
df2 = spark.createDataFrame(l2) 

df1 :

["hello","world"]
["stack","overflow"]
["hello","alice"]
["sample","text"]

df2 :

["big","world"]
["sample","overflow","alice","text","bob"]
["hello", "sample"]

Pour chaque ligne de df1, je veux calculer le nombre de fois que tous les mots du tableau apparaissent dans df2.

Par exemple, la première ligne de df1 est la suivante ["hello","world"] . Maintenant, je veux vérifier que df2 est l'intersection de ["hello","world"] avec chaque ligne de df2.

|                  ARRAY                    | INTERSECTION | LEN(INTERSECTION)| 
|["big","world"]                            |["world"]     | 1                |
|["sample","overflow","alice","text","bob"] |[]            | 0                |   
|["hello","sample"]                         |["hello"]     | 1                |

Maintenant, je veux retourner le sum(len(interesection)) . Au final, je veux que le df1 résultant ressemble à ceci :

résultat df1 :

       ARRAY               INTERSECTION_TOTAL
| ["hello","world"]    |      2                 |
| ["stack","overflow"] |      1                 |
| ["hello","alice"]    |      2                 |
| ["sample","text"]    |      3                 |

Comment puis-je résoudre ce problème ?

Answer 1

Je me concentrerais d'abord sur l'évitement du produit cartésien. J'essaierais d'exploser et de rejoindre

from pyspark.sql.functions import explode, monotonically_increasing_id

df1_ = (df1.toDF("words")
  .withColumn("id_1", monotonically_increasing_id())
  .select("*", explode("words").alias("word")))

df2_ = (df2.toDF("words")
    .withColumn("id_2", monotonically_increasing_id())
    .select("id_2", explode("words").alias("word")))

(df1_.join(df2_, "word").groupBy("id_1", "id_2", "words").count()
    .groupBy("id_1", "words").sum("count").drop("id_1").show())
+-----------------+----------+                                                  
|            words|sum(count)|
+-----------------+----------+
|   [hello, alice]|         2|
|   [sample, text]|         3|
|[stack, overflow]|         1|
|   [hello, world]|         2|
+-----------------+----------+

Si les valeurs intermédiaires ne sont pas nécessaires, on peut les simplifier :

df1_.join(df2_, "word").groupBy("words").count().show()
+-----------------+-----+                                                       
|            words|count|
+-----------------+-----+
|   [hello, alice]|    2|
|   [sample, text]|    3|
|[stack, overflow]|    1|
|   [hello, world]|    2|
+-----------------+-----+

et vous pouvez omettre d'ajouter des identifiants.