问题标签 [gelly]

我正在尝试将 Gelly Graph 扩展到带有 Tuple5 而不是 Tuple 3 的边缘的 Graph。通过扩展 gelly Graph 是不可能的，因为构造函数是私有的。我继续制作了自己的 Edge 类，它扩展了 Tuple5 而不是 Tuple 3，但我不确定此时我是否可以轻松地将它变成一个 gelly Graph，因为 API 指向边集的 tuple3 或 tuple2。

问题是这是否可以更容易地完成，或者我想做的事情是否有可能

在 Gelly 中，我正在尝试制作一个称为 Temporal edge 的特殊 Edge，为了使这更容易，我制作了一个名为 Temporaledgev3 的类：

现在我正在尝试将这些边添加到 Flink DataSet 中，以便它可以在图中使用，但我似乎无法弄清楚如何。但是，当我使用具有相同构造函数的 Edge 类时，它可以工作。

这是代码，最后一行给出了错误

错误：错误的第一个参数类型。找到：'org.apache.flink.api.java.DataSet>'，需要：'org.apache.flink.api.java.DataSet>' 少...

也许我只是不知道如何使用泛型类型

我的目标是在Flink Gelly中实现接近中心性，并尝试在每个节点上使用 map 函数来计算 SSSP。图形未序列化，因此我无法将其发送到每个节点。任何帮助将不胜感激。

我正在尝试在 Flink 中编写一个需要两个阶段的计算。

在第一阶段，我创建一个 Graph 并获取它的顶点 ID：

在第二阶段，我想使用这些 id 为每个顶点运行 SingleSourceShortestPath。

它在本地工作（在 IntelliJ IDE 和命令行中使用./bin/flink run ...），但是当我使用其 WebUI 在 Flink 上提交作业时，程序只执行 untilcollect()方法并且不运行程序的其余部分（对于语句和print()）。

问题是什么？

这是我的代码：

我一直在试验 Flink 流式传输，使用 Yahoo 流式传输基准测试之类的基准测试：https ://github.com/yahoo/streaming-benchmarks这应该会给系统带来压力，但我从未达到令人满意的 CPU 利用率 - 在事实上，使用所有可用系统内核（并行度 = 节点*内核）和每个内核一个 TaskManager 插槽时，它主要低至 ~25%。

最近，我开始使用 Gelly，Flink 的 Graph API 使用一些提供的示例算法（例如 Pagerank），批处理从数万到数亿个顶点的数据集。

我占用了四个 TaskManager，每个具有 32 个内核，并且按照我设置的文档的建议taskmanager.numberOfTaskSlots: 32和parallelism.default: 128.

即使我增加这些值，平均 CPU 利用率也永远不会超过 40%。因此，由于我的资源没有得到充分利用，我的绩效很低。

我还想指出这样一个事实，在某些情况下，我注意到在较低的并行度（和 CPU 利用率）下性能更好。

我错过了什么？

我注意到 Flink Gelly 性能的奇怪行为（PageRank 算法在大约 17,000,000 个顶点和 136,000,000 个边的图上）。

虽然将taskmanager.heap.mb: 125000结果设置为 ~110 秒的运行时间，taskmanager.heap.mb: 10000但运行时间减少了 50% 以上。

我知道大型 JVM 需要很多时间来初始化，垃圾收集也很慢，但是这些原因足以导致性能如此大的差异吗？

Flink 图形库 Gelly 可以做图划分、图模式匹配、连接等吗？

我运行具有以下规格的集群（每个任务管理器）： - 16 个 CPU 线程 - 16 GB 内存 - 16 个插槽。我有两个任务管理器，当我们运行连接组件等图形算法时，程序将失败并出现以下错误：

线程“主”org.apache.flink.runtime.client.JobExecutionException 中的异常：java.lang.RuntimeException：内存用完。压缩失败。numPartitions：32 minPartition：29 maxPartition：30 溢出段数：105 bucketSize：234 总内存：42598400 分区内存：30539776 消息：org.apache.flink.runtime.minicluster.MiniCluster.executeJobBlocking（MiniCluster.java:625）处为空在 org.apache.flink.client.LocalExecutor.executePlan(LocalExecutor.java:234) 在 org.apache.flink.api.java.LocalEnvironment.execute(LocalEnvironment.java:91) 在 org.apache.flink.api.java .ExecutionEnvironment.execute(ExecutionEnvironment.java:816) at org.apache.flink.api.java.DataSet.collect(DataSet.java:413) at org.apache.flink.api.java.DataSet.print(DataSet.java :1652) 在 com.asha.adw.ga.gpe.main。Main.main(Main.java:207) 引起：java.lang.RuntimeException：内存用完。压缩失败。numPartitions：32 minPartition：29 maxPartition：30 溢出段数：105 bucketSize：234 总内存：42598400 分区内存：30539776 消息：org.apache.flink.runtime.operators.hash.CompactingHashTable.insertRecordIntoPartition(CompactingHashTable.java: 457）在org.apache.flink.runtime.operators.hash.CompactingHashTable.buildTableWithUniqueKey（CompactingHashTable.java:316）在org.apache.flink. .apache.flink.runtime.iterative.task.IterationHeadTask.readInitialSolutionSet(IterationHeadTask.java:228) 在 org.apache.flink.runtime.iterative.task.IterationHeadTask.run(IterationHeadTask.java:

我监控了堆，它的利用率约为 40%。我也设置了off-heap true，但是没有这样的运气。当将主内存增加到 64 GB 时，它成功运行。任何帮助将不胜感激。

我设置了一个 Flink 集群，配置如下：

任务管理器数量：2

org.apache.flink.graph.library.ConnectedComponents当我在具有 400 万条边和 1,750,000 个顶点且并行度为 2 的图上运行连接组件算法 ( ) 时，我得到了这个异常：

java.lang.RuntimeException：获取排序输入时出错：线程“SortMerger Reading Thread”因异常而终止：记录超出排序缓冲区的最大大小（当前最大值：87949312 字节）。在 org.apache.flink.runtime.operators.sort.UnilateralSortMerger.getIterator(UnilateralSortMerger.java:619) 在 org.apache.flink.runtime.operators.BatchTask.getInput(BatchTask.java:1108) 在 org.apache.flink .runtime.operators.NoOpDriver.run(NoOpDriver.java:82) 在 org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:503) 在 org.apache.flink.runtime.operators.BatchTask.invoke (BatchTask.java:368) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703) at java.lang.Thread.run(Thread.java:748) 原因：java.io。 IOException: 线程 'SortMerger 读取线程' 由于异常而终止：记录超出排序缓冲区的最大大小（当前最大值：87949312 字节）。at org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:800) 原因：java.io.IOException：记录超出排序缓冲区的最大大小（当前最大值：87949312字节）。在 org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ReadingThread.go(UnilateralSortMerger.java:955) 在 org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:796) 记录超过了排序缓冲区的最大大小（当前最大值：87949312 字节）。在 org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ReadingThread.go(UnilateralSortMerger.java:955) 在 org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:796) 记录超过了排序缓冲区的最大大小（当前最大值：87949312 字节）。在 org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ReadingThread.go(UnilateralSortMerger.java:955) 在 org.apache.flink.runtime.operators.sort.UnilateralSortMerger$ThreadBase.run(UnilateralSortMerger.java:796)

根据我的配置，我不知道为什么它说The record exceeds the maximum size of a sort buffer (current maximum:87949312 bytes)有足够的内存段。

有谁知道如何解决这个问题？

我是 Fink 和 Gelly 的新手。我们需要进行图形计算，而图形本身由于增量更新（经常）而不断变化。因为图太大了，我们用 TinkerPop 对 Spark 做了一些调查。但是这种方法的问题是RDD是不可变的，这使得图上的增量更新有点难以处理（IndexRDD已经很久没有维护了，所以我们不想往这个方向发展）。这让我们想到了 Flink 和 Gelly。任何专家可以帮我回答这些问题： 1. Gelly/Flink 可以原生处理增量更新，对吗？我的意思是图遍历->图变异->图遍历不需要像spark这样的额外努力来重新加载整个RDD，对吧？但是如果是这样，那么在 Flink 中是如何保持数据一致性的呢？2.

非常感谢