spark edge validation

spark_edge_validation.scala

val oid_info = Seq((1,1), (1, 2), (1, 1), (2, 1), (2, 3), (3, 4), (3, 3), (3, 4)).toDF("oid", "ip")

"""
+---+---+
|oid| ip|
+---+---+
|  1|  1|
|  1|  2|
|  1|  1|
|  2|  1|
|  2|  3|
|  3|  4|
|  3|  3|
|  3|  4|
+---+---+
"""

val edges = Seq((1, 2), (2, 1), (3, 2), (2, 3), (1, 3), (3, 1)).toDF("u", "v")

"""
+---+---+
|  u|  v|
+---+---+
|  1|  2|
|  2|  1|
|  3|  2|
|  2|  3|
|  1|  3|
|  3|  1|
+---+---+
"""



val fat = edges.join(oid_info.selectExpr("oid as u_oid", "ip as u_ip"), col("u") === col("u_oid")).join(oid_info.selectExpr("oid as v_oid", "ip as v_ip"), col("v") === col("v_oid"))


"""
+---+---+-----+----+-----+----+
|  u|  v|u_oid|u_ip|v_oid|v_ip|
+---+---+-----+----+-----+----+
|  1|  3|    1|   1|    3|   4|
|  1|  3|    1|   1|    3|   3|
|  1|  3|    1|   1|    3|   4|
|  1|  2|    1|   1|    2|   3|
|  1|  2|    1|   1|    2|   1|
|  1|  3|    1|   2|    3|   4|
|  1|  3|    1|   2|    3|   3|
|  1|  3|    1|   2|    3|   4|
|  1|  2|    1|   2|    2|   3|
|  1|  2|    1|   2|    2|   1|
|  1|  3|    1|   1|    3|   4|
|  1|  3|    1|   1|    3|   3|
|  1|  3|    1|   1|    3|   4|
|  1|  2|    1|   1|    2|   3|
|  1|  2|    1|   1|    2|   1|
|  2|  3|    2|   1|    3|   4|
|  2|  3|    2|   1|    3|   3|
|  2|  3|    2|   1|    3|   4|
|  2|  1|    2|   1|    1|   1|
|  2|  1|    2|   1|    1|   2|
+---+---+-----+----+-----+----+
only showing top 20 rows
"""

fat.groupBy("u", "v").agg(count(when(col("u_ip") === col("v_ip"), 1)).as("count")).filter(col("count") < 1).show

"""
+---+---+-----+
|  u|  v|count|
+---+---+-----+
|  3|  1|    0|
|  1|  3|    0|
+---+---+-----+
"""