apache spark - ¿Cómo agrupar por elemento común en la matriz?

apache-spark apache-spark-sql

Question

Estoy tratando de encontrar la solución en spark para agrupar datos con un elemento común en una matriz.

 key value
[k1,k2] v1
[k2] v2
[k3,k2] v3
[k4] v4

Si algún elemento coincide en la clave, tenemos que asignar el mismo groupid a eso. (Grupo por elemento común)

Resultado:

key value GroupID
[k1,k2] v1 G1
[k2] v2 G1
[k3,k2] v3 G1
[k4] v4 G2

Ya se dan algunas sugerencias con Spark Graphx, pero en este momento la curva de aprendizaje será más para implementar esto para una sola característica.

Answer 1

Incluya graphframes (la última versión compatible de Spark es 2.1, pero también debería ser compatible con 2.2, si usa una más nueva, tendrá que construir la suya con el parche 2.3) reemplazando XXX con la versión Spark e YYY con la versión Scala:

spark.jars.packages graphframes:graphframes:0.5.0-sparkXXX-s_YYY

Añadir claves de explosión:

import org.apache.spark.sql.functions._
val df = Seq(
(Seq("k1", "k2"), "v1"), (Seq("k2"), "v2"),
(Seq("k3", "k2"), "v3"), (Seq("k4"), "v4")
).toDF("key", "value")
val edges = df.select(
explode($"key") as "src", $"value" as "dst")

Convierte a graphframe :

import org.graphframes._
val gf = GraphFrame.fromEdges(edges)

Establecer el directorio del punto de control (si no está configurado):

import org.apache.spark.sql.SparkSession
val path: String = ???
val spark: SparkSession = ???
spark.sparkContext.setCheckpointDir(path)

Encuentra componentes conectados:

val components = GraphFrame.fromEdges(edges).connectedComponents.setAlgorithm("graphx").run

Unir resultado con datos de entrada:

 val result = components.where($"id".startsWith("v")).toDF("value", "group").join(df, Seq("value"))

Comprobar resultado:

result.show
// +-----+------------+--------+
// |value| group| key|
// +-----+------------+--------+
// | v3|489626271744|[k3, k2]|
// | v2|489626271744| [k2]|
// | v4|532575944704| [k4]|
// | v1|489626271744|[k1, k2]|
// +-----+------------+--------+