标签: spark-graphx

刚开始我的图表处理方法和工具的游览.我们基本上做了什么 - 计算一些标准指标,如pagerank,聚类系数,三角计数,直径,连通性等.过去对Octave感到满意,但是当我们开始使用图表时让我们说10 ^ 9个节点/边缘我们卡住了.

因此,可能的解决方案可以是使用Hadoop/Giraph,Spark/GraphX,Neo4j在其上的分布式云等.

但由于我是初学者,有人可以建议实际选择什么？在使用Spark/GraphX和Neo4j时,我没有什么区别？现在我考虑Spark/GraphX,因为它有更多类似Python的语法,而neo4j有自己的Cypher.neo4j中的可视化很酷但在如此大规模上没有用.我不明白是否有理由使用额外级别的软件(neo4j)或只使用Spark/GraphX？因为我理解neo4j不会节省那么多时间,就像我们使用纯粹的hadoop vs Giraph或GraphX或Hive一样.

谢谢.

我有一个由两个工作节点组成的集群.Worker_Node_1 - 64GB RAM Worker_Node_2 - 32GB RAM

背景摘要: 我正在尝试在纱线簇上执行spark-submit以在Graph上运行Pregel来计算从一个源顶点到所有其他顶点的最短路径距离,并在控制台上打印这些值.实验:

1. 对于具有15个顶点的小图,执行完成应用程序最终状态:SUCCEEDED
2. 我的代码完美地工作并打印单个顶点作为源顶点的241个顶点图的最短距离,但是存在问题.

问题: 当我深入了解日志文件时,任务在4分钟和26秒内成功完成,但仍然在终端上它继续显示应用程序状态为运行,大约12分钟后任务执行终止说 -

Application application_1447669815913_0002 failed 2 times due to AM Container for appattempt_1447669815913_0002_000002 exited with exitCode: -104 For more detailed output, check application tracking page:http://myserver.com:8088/proxy/application_1447669815913_0002/
Then, click on links to logs of each attempt. 
Diagnostics: Container [pid=47384,containerID=container_1447669815913_0002_02_000001] is running beyond physical memory limits. Current usage: 17.9 GB of 17.5 GB physical memory used; 18.7 GB of 36.8 GB virtual memory used. Killing container. …

GraphX附带了一种用于查找图形的连通分量的算法.

我没有找到关于其实现复杂性的声明.

通常,查找连通分量可以在线性时间内完成,例如通过广度优先搜索或深度优先搜索(参见维基百科文章).但是,假设您可以将图形保留在内存中.GraphX实现了一个分布式的核外算法,因此我认为它不具有可比性.

你知道他们的算法如何工作以及它有多复杂性？

我正在尝试使用Spark的GraphX库实现拓扑排序.

这是我到目前为止编写的代码:

MyObject.scala

import java.util.ArrayList

import scala.collection.mutable.Queue

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.graphx.Edge
import org.apache.spark.graphx.EdgeDirection
import org.apache.spark.graphx.Graph
import org.apache.spark.graphx.Graph.graphToGraphOps
import org.apache.spark.graphx.VertexId
import org.apache.spark.rdd.RDD
import org.apache.spark.rdd.RDD.rddToPairRDDFunctions

object MyObject {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Spark-App").setMaster("local[2]")
    val sc = new SparkContext(conf)

    val resources: RDD[Resource] = makeResources(sc)
    val relations: RDD[Relation] = makeRelations(sc)

    println("Building graph ...")
    var graph = buildGraph(resources, relations, sc)
    println("Graph built!!")

    println("Testing topo sort ...")
    val topoSortResult = topoSort(graph, sc);
    println("topoSortResult = …

我有下面的节点和边缘给出的有向图.

节点

1,2,3,4,5

边缘

(1,2),(1,3),(1,4),(2,5),(3,4),(3,5),(4,5)

如何将此有向图转换为无向图我是否必须使用内置方法进行转换.如果有内置方法,它是什么方法？或者,我是否必须在数据集中手动添加边,例如(1,2)到(2,1).

我在MAC中运行Bash脚本.此脚本调用以Scala语言编写的spark方法很多次.我目前正在尝试使用for循环调用此spark方法100,000次.

在运行少量迭代(大约3000次迭代)之后,代码以以下异常退出.

org.apache.spark.rpc.RpcTimeoutException: Futures timed out after [10 seconds]. This timeout is controlled by spark.executor.heartbeatInterval
    at org.apache.spark.rpc.RpcTimeout.org$apache$spark$rpc$RpcTimeout$$createRpcTimeoutException(RpcTimeout.scala:48)
    at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:63)
    at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:59)
    at scala.PartialFunction$OrElse.apply(PartialFunction.scala:167)
    at org.apache.spark.rpc.RpcTimeout.awaitResult(RpcTimeout.scala:83)
    at org.apache.spark.rpc.RpcEndpointRef.askWithRetry(RpcEndpointRef.scala:102)
    at org.apache.spark.executor.Executor.org$apache$spark$executor$Executor$$reportHeartBeat(Executor.scala:518)
    at org.apache.spark.executor.Executor$$anon$1$$anonfun$run$1.apply$mcV$sp(Executor.scala:547)
    at org.apache.spark.executor.Executor$$anon$1$$anonfun$run$1.apply(Executor.scala:547)
    at org.apache.spark.executor.Executor$$anon$1$$anonfun$run$1.apply(Executor.scala:547)
    at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1877)
    at org.apache.spark.executor.Executor$$anon$1.run(Executor.scala:547)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304)

Exception in thread "dag-scheduler-event-loop" 16/11/22 13:37:32 WARN NioEventLoop: Unexpected exception in the selector loop.
java.lang.OutOfMemoryError: Java heap space
    at io.netty.util.internal.MpscLinkedQueue.offer(MpscLinkedQueue.java:126)
    at io.netty.util.internal.MpscLinkedQueue.add(MpscLinkedQueue.java:221)
    at io.netty.util.concurrent.SingleThreadEventExecutor.fetchFromScheduledTaskQueue(SingleThreadEventExecutor.java:259)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:346)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:357)
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
    at java.lang.Thread.run(Thread.java:745)
java.lang.OutOfMemoryError: Java heap space
    at …

如何使用子图函数来获取仅包含特定连接组件的顶点和边的图形？假设我知道连接组件id,最终目标是基于连接组件创建新图.我想保留原始图形中的顶点属性.

我正在寻找一种方法来可视化Spark的Graphx中构建的图形.据我所知,Graphx没有任何可视化方法,因此我需要将Graphx中的数据导出到另一个图库,但我被困在这里.我遇到了这个网站:https://lintool.github.io/warcbase-docs/Spark-Network-Analysis/ 但它没有帮助.我应该使用哪个库以及如何导出图表.

我遇到了Out of Memeory错误,原因不明,我已经立即释放了无用的RDD,但经过几轮循环后,OOM错误仍然出现.我的代码如下:

// single source shortest path
def sssp[VD](graph:Graph[VD,Double], source: VertexId): Graph[Double, Double] = {
    graph.mapVertices((id, _) => if (id == source) 0.0 else Double.PositiveInfinity)
        .pregel(Double.PositiveInfinity)(
            (id, dist, newDist) => scala.math.min(dist, newDist),
            triplet => {
                if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
                    Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
                }
                else {
                    Iterator.empty
                }
            },
            (a, b) => math.min(a, b)
        )
}

def selectCandidate(candidates: RDD[(VertexId, (Double, Double))]): VertexId = {
    Random.setSeed(System.nanoTime())
    val selectLow = Random.nextBoolean()
    val (vid, (_, _)) = if (selectLow) …

我正在尝试计算spark graphx图中节点值的总和.简而言之,图形是一棵树,顶部节点(根)应该是所有孩子和他们的孩子的总和.我的图实际上是一棵看起来像这样的树,预期的总和值应该是1850:

                                     +----+
                     +--------------->    |  VertexID 14
                     |               |    |  Value: 1000
                 +---+--+            +----+
    +------------>      | VertexId 11
    |            |      | Value:     +----+
    |            +------+ Sum of 14 & 24  |  VertexId 24
+---++                +-------------->    |  Value: 550
|    | VertexId 20                   +----+
|    | Value:
+----++Sum of 11 & 911
      |
      |           +-----+
      +----------->     | VertexId 911
                  |     | Value: 300
                  +-----+

对此的第一次尝试如下:

val vertices: RDD[(VertexId, Int)] =
      sc.parallelize(Array((20L, 0)
        , (11L, 0)
        , (14L, 1000)
        , …