Spark Dataset - NullPointerException en faisant un filtre sur l'ensemble de données

Question

J'ai 2 jeux de données avec moi comme montré ci-dessous. J'essaie de savoir combien de produits sont associés à chaque jeu. Fondamentalement, j'essaie de garder un compte du nombre de produits associés.

scala> df1.show() 
    gameid  | games  | users   | cnt_assoc_prod 
    ------------------------------------------- 
    1   | cricket |[111, 121]  | 
    2   | basketball|[211]   | 
    3   | skating |[101, 100, 98] | 

    scala> df2.show() 
    user | products 
    ---------------------- 
    98 | "shampoo" 
    100 | "soap"   
    101 | "shampoo" 
    111 | "shoes" 
    121 | "honey" 
    211 | "shoes" 

Je suis en train de parcourir à travers chacun des utilisateurs de DF1 du tableau et trouver la ligne correspondante dans DF2 en appliquant le filtre sur la colonne correspondant à l'utilisateur.

df1.map{x => { 
    var assoc_products = new Set() 
    x.users.foreach(y => assoc_products + df2.filter(z => z.user == y).first(). 
     products) 
    x.cnt_assoc_prod = assoc_products.size 
} 

Tout en application du filtre on me suit Exception

java.lang.NullPointerException 
    at org.apache.spark.sql.Dataset.logicalPlan(Dataset.scala:784) 
    at org.apache.spark.sql.Dataset.mapPartitions(Dataset.scala:344) 
    at org.apache.spark.sql.Dataset.filter(Dataset.scala:307) 

J'utilise la version 1.6.1 d'allumage.

Answer 1

Vous pouvez exploser la colonne users dans df1, se joindre à df2 sur la colonne user, puis faire le groupBy comte:

(df1.withColumn("user", explode(col("users"))) 
    .join(df2, Seq("user")) 
    .groupBy("gameid", "games") 
    .agg(count($"products").alias("cnt_assoc_prod")) 
).show 

+------+----------+--------------+ 
|gameid|  games|cnt_assoc_prod| 
+------+----------+--------------+ 
|  3| skating|    3| 
|  2|basketball|    1| 
|  1| cricket|    2| 
+------+----------+--------------+