使用 Spark Graphframes 进行分区答案

【问题标题】：Partitioning with Spark Graphframes使用 Spark Graphframes 进行分区
【发布时间】：2019-11-27 09:39:52
【问题描述】：

我正在使用 Spark Graphframes 处理一个较大的 (?) 图（6000 万个顶点和 95 亿条边）。基础数据并不大——顶点在磁盘上占用大约 500mb，边大约是 40gb。由于 java 堆内存不足的问题，我的容器经常关闭，但我认为根本问题是 graphframe 不断地洗牌数据（我看到洗牌读/写高达 150gb）。有没有办法有效地划分 Graphframe 或底层边/顶点以减少 shuffle？

【问题讨论】：

标签： apache-spark graphframes

【解决方案1】：

TL;DR 无法有效分区Graphframe。

Graphframe算法可以分为两类：

将处理委托给GraphX 对应方的方法。 GraphX 支持多种分区方法，但这些方法并未通过Graphframe API 公开。如果您使用其中之一，最好直接使用GraphX。

不幸的是，GraphX 的开发在过去两年中几乎完全停止，只有少数几个小修复，与核心库和out-of-core libraries 相比，整体性能非常令人失望。
使用 SparkDatasets 原生实现的方法，考虑到有限的编程模型和只有单一的分区模式，非常不适合复杂的图形处理。

虽然关系列式存储可用于高效的图形处理，但 Graphframes 采用的朴素迭代 join 方法只是不能扩展（但对于一跳或两跳的浅层遍历是可以的）。'

您可以尝试将vertices 和edges DataFrames 分别重新分区为id 和src：
```
val nPart: Int = ???

GraphFrame(v.repartition(nPart, v("id")), e.repartition(e(nPart, "src")))
```
在某些情况下应该有什么帮助。

总体而言，就目前（2016 年 12 月）的状态而言，Spark 不是密集图分析的好选择。

【讨论】：

欣赏洞察力。我发现我可以通过使用 GraphX partition schema 在边缘数据框中创建一个自定义列并对其进行分区来提高性能。
@John 我想不通，你有关于如何对自定义列进行分区的示例吗？
@John 你能分享一下你是如何解决这个性能问题的更新吗？
@John +1 获取描述您如何实现分区架构的编码示例

【解决方案2】：

这是部分解决方案/解决方法 - 创建一个模拟分区函数之一的 UDF，以在其上创建一个新列和分区。

num_parts = 256
random_vertex_cut = udf.register("random_vertex_cut", lambda src, dst: math.abs((src, dst).hashCode()) % num_parts, IntegerType())

edge.withColumn("v_cut", random_vertex_cut(col("src"), col("dst")).repartition(256, "v_cut")

这种方法可以帮助一些人，但不如 GraphX。

【讨论】：