Доступ к атрибутам ребер из кортежа в GraphX

Question

У меня есть другая проблема с моим графиком. Мои ребра содержат три атрибута в виде кортежа, поэтому ребро выглядит как (verticeId_1, verticeId_2,(Attr1,Attr2,Attr3)). Я пытаюсь создать подграф, содержащий только ребра, где Attr2> 100 (Attr2 это число). Поэтому я пробую это: val gr2=graph.subgraph(epred=(id1: String,id2: String, attr: Tuple) => attr(2)>100) , но становиться ошибкой Кортежи нельзя напрямую деструктурировать в параметрах метода или функции . Есть ли какое-то решение этой проблемы? Заранее спасибо.

UPDATE. Края определены как val ed = df.rdd.map(x =>((MurmurHash.stringHash(x(3).toString),MurmurHash.stringHash(x(4).toString)), (x(0),x(1),x(2)))).map(x => Edge(x._1._1, x._1._2,x._2)) из кадра данных df, который содержит атрибуты в виде столбцов 0,1,2 и вершин в качестве столбцов 3 и 4.

Answer 1

Функция подграфа может принимать предикаты вершин и ребер. Для предиката ребра требуется функция, которая принимает объект EdgeTriplet и выдает логическое значение. У объекта EdgeTriple есть атрибуты, которые дадут вам доступ к информации об обеих вершинах и об атрибутах созданной вами ссылки. Чтобы реализовать ваш случай, попробуйте это:

g2 = graph.subgraph( epred=(triplet) => triplet.attr._3 > 100)

Вот документация для EdgeTriple, которая показывает, какие атрибуты у вас есть:

https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.graphx.EdgeTriplet

Кроме того, обратите внимание, что ваши атрибуты ребер являются кортежем, поэтому они должны быть доступны с использованием таких элементов, как "_1", "_2" и т. Д. Вместо индексации массива с использованием скобок.

РЕДАКТИРОВАТЬ: Для справки, вот полный автономный пример:

import org.apache.spark._
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
import scala.util.hashing.MurmurHash3

val data = Seq(
    (10, 11, 12, 1, 2),
    (10, 11, 13, 1, 3),
    (10, 11, 14, 1, 4),
    (10, 11, 15, 2, 3),
    (10, 11, 16, 2, 4)
)

val df = sc.parallelize(data).toDF


val ed = df.rdd.
            map(x =>((MurmurHash3.stringHash(x.getString(3)), MurmurHash3.stringHash(x.getString(4))), (x.getInt(0), x.getInt(1), x.getInt(2)))).
            map(x => Edge(x._1._1, x._1._2, x._2))

val graph = Graph.fromEdges(ed, 0)

graph.subgraph( epred=(triplet) => triplet.attr._3 > 5)