Spark: как отфильтровать данные по условию подмножества - PullRequest
Новая
       1

Spark: как отфильтровать данные по условию подмножества

0 голосов
/ 16 октября 2019

У меня есть две таблицы, отображение p_to_v, отображение g_to_v. 

scala> val p_to_v = Seq(("p1", "v1"), ("p1", "v2"), ("p2", "v1")).toDF("p", "v")
scala> p_to_v.show
+---+---+
|  p|  v|
+---+---+
| p1| v1|
| p1| v2|
| p2| v1|
+---+---+

«p1» отображается на [v1, v2]

«p2» отображается на [v1] 

scala> val g_to_v = Seq(("g1", "v1"), ("g2", "v1"), ("g2", "v2"), ("g3", "v2")).toDF("g", "v")
scala> g_to_v.show
+---+---+
|  g|  v|
+---+---+
| g1| v1|
| g2| v1|
| g2| v2|
| g3| v2|
+---+---+

"g1" отображается на [v1]

"g2" отображается на [v1, v2]

"g3" отображается на [v2]

Я хочу получить все комбинации p и g , для которых соответствующее отображение v p является подмножеством отображения v г 

p1 [v1, v2] - g2 [v1, v2]
p2 [v1] - g1 [v1]
p2 [v1] - g2 [v1, v2]

Как я могу получить то же самое?

1 Ответ

1 голос
/ 16 октября 2019

Это довольно просто. Вам нужно использовать groupBy, а затем простое внутреннее объединение 

scala> val p_to_v = Seq(("p1", "v1"), ("p1", "v2"), ("p2", "v1")).toDF("p", "v")
19/10/16 22:11:55 WARN metastore: Failed to connect to the MetaStore Server...
p_to_v: org.apache.spark.sql.DataFrame = [p: string, v: string]

scala> val g_to_v = Seq(("g1", "v1"), ("g2", "v1"), ("g2", "v2"), ("g3", "v2")).toDF("g", "v")
g_to_v: org.apache.spark.sql.DataFrame = [g: string, v: string]

Теперь выполните операцию groupBy 

scala> val pv = p_to_v.groupBy($"p").agg(collect_list("v").as("pv"))
pv: org.apache.spark.sql.DataFrame = pv = [p: string, pv: array<string>]

scala> val gv = g_to_v.groupBy($"g").agg(collect_list("v").as("gv"))
gv: org.apache.spark.sql.DataFrame = [g: string, gv: array<string>]

scala> pv.show
+---+--------+
|  p|      pv|
+---+--------+
| p2|    [v1]|
| p1|[v1, v2]|
+---+--------+

scala> gv.show
+---+--------+
|  g|      gv|
+---+--------+
| g2|[v2, v1]|
| g3|    [v2]|
| g1|    [v1]|
+---+--------+

Создайте UDF для поиска подмножества 

import org.apache.spark.sql.Row
import org.apache.spark.sql.functions._

def subLis(ar1: Seq[Row], ar2: Seq[Row]) = ar1.toSet.subsetOf(ar2.toSet)
    subLis: (ar1: Seq[org.apache.spark.sql.Row], ar2: Seq[org.apache.spark.sql.Row])Boolean

val subLisUDF = udf(subLis _)
    UserDefinedFunction(<function2>,BooleanType,None)

Теперь выможет выполнить перекрестное соединение и применить UDF 

spark.conf.set("spark.sql.crossJoin.enabled", "true")
pv.join(gv).withColumn("newdsa", subLisUDF($"pv", $"gv")).filter($"newdsa").show

+---+--------+---+--------+------+
|  p|      pv|  g|      gv|newdsa|
+---+--------+---+--------+------+
| p2|    [v1]| g2|[v2, v1]|  true|
| p1|[v1, v2]| g2|[v2, v1]|  true|
| p2|    [v1]| g1|    [v1]|  true|
+---+--------+---+--------+------+

или объединиться с условием 

pv.join(gv, pv("pv") === gv("gv") || subLisUDF($"pv", $"gv")).show

+---+--------+---+--------+
|  p|      pv|  g|      gv|
+---+--------+---+--------+
| p2|    [v1]| g2|[v1, v2]|
| p1|[v1, v2]| g2|[v1, v2]|
| p2|    [v1]| g1|    [v1]|
+---+--------+---+--------+

Попробуйте оба варианта и выберите наиболее эффективный.

...