我是GraphX的新手,有一个Spark数据框,其中包含四列,如下所示:
src_ip dst_ip flow_count sum_bytes
8.8.8.8 1.2.3.4 435 1137
... ... ... ...
基本上,我想同时映射src_ip
和dst_ip
顶点,并指定flow_count
和sum_bytes
作为edges属性。据我所知,我们不能在GraphX中添加edge属性,因为只允许顶点属性。因此,我正在考虑添加flow_count
作为边缘权重:
//create edges
val trafficEdges = trafficsFromTo.map(x =Edge(MurmurHash3.stringHash(x(0).toString,MurmurHash3.stringHash(x(1).toString,x(2))
但是,我也可以添加sum_bytes
边缘权重吗?
可以将两个变量都添加到边缘。最简单的解决方案是使用元组,例如:
val data = Array(Edge(3L, 7L, (123, 456)), Edge(5L, 3L, (41, 34)))
val edges: RDD[Edge[(Int, Int)]] = spark.sparkContext.parallelize(data)
另外,您可以使用案例类:
case class EdgeWeight(flow_count: Int, sum_bytes: Int)
val data2 = Array(Edge(3L, 7L, EdgeWeight(123, 456)), Edge(5L, 3L, EdgeWeight(41, 34)))
val edges: RDD[Edge[EdgeWeight]] = spark.sparkContext.parallelize(data2)
如果要添加更多属性,使用案例类将更方便使用和维护。
我认为,在这种特定情况下,可以通过以下方法最优雅地解决:
val trafficEdges = trafficsFromTo.map{x =>
Edge(MurmurHash3.stringHash(x(0).toString,
MurmurHash3.stringHash(x(1).toString,
EdgeWeight(x(2), x(3))
}
trafficEdges.sortBy(edge => edge.attr.flow_count) // sort by flow_count
本文收集自互联网,转载请注明来源。
如有侵权,请联系 [email protected] 删除。
我来说两句