4. Spark GraphX 入门：核心概念、RDD 与 PropertyGraph、创建 Graph 的三种方式

在欺诈风控场景中，实体（用户、设备、账户）与关系（交易、登录、担保）天然构成一张异构图。Spark GraphX 是 Spark 生态中专门用于图并行计算的 API，它基于 RDD 构建，提供了统一的属性图模型。本节将系统讲解 GraphX 的核心概念、RDD 与 PropertyGraph 的关系，以及创建 Graph 的三种标准方式。

4.1 GraphX 核心概念

GraphX 将图抽象为 PropertyGraph，即带属性的有向多重图。其核心由两个 RDD 构成：

4.2 RDD 与 PropertyGraph 的关系

PropertyGraph 本质上是一个 逻辑容器，它同时持有 vertices 和 edges 两个 RDD 的引用。GraphX 提供的所有图算子（如 mapVertices、aggregateMessages）最终都会转化为对这两个 RDD 的转换操作。

下图展示了 RDD 与 PropertyGraph 之间的映射关系：

4.3 创建 Graph 的三种方式

在实际欺诈风控项目中，数据源通常来自 Hive 表、Kafka 流或 Parquet 文件。GraphX 提供了三种标准创建方式，以适应不同场景。

方式一：从 RDD 直接构建

最通用的方式，分别构造顶点 RDD 和边 RDD，然后调用 Graph(vertices, edges) 或 Graph.fromEdges。

import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD

// 1. 构造顶点 RDD: (VertexId, 属性)
val vertices: RDD[(VertexId, (String, Int))] = sc.parallelize(Seq(
  (1L, ("Alice", 0)),   // 0: 正常用户
  (2L, ("Bob", 1)),     // 1: 欺诈用户
  (3L, ("Charlie", 0))
))

// 2. 构造边 RDD: Edge(源ID, 目标ID, 属性)
val edges: RDD[Edge[Double]] = sc.parallelize(Seq(
  Edge(1L, 2L, 5000.0),  // 转账 5000 元
  Edge(2L, 3L, 200.0),
  Edge(1L, 3L, 1000.0)
))

// 3. 创建属性图
val graph: Graph[(String, Int), Double] = Graph(vertices, edges)

// 4. 查看顶点数
println(s"顶点数: ${graph.vertices.count()}")   // 输出: 3
println(s"边数: ${graph.edges.count()}")        // 输出: 3

方式二：从边 RDD 自动推断顶点

当只有边数据，且顶点属性可以后续补充时，使用 Graph.fromEdges。顶点属性会被设为默认值（通常为 1）。

// 只有边数据，顶点自动从边的两端推断
val edgesOnly: RDD[Edge[String]] = sc.parallelize(Seq(
  Edge(1L, 2L, "转账"),
  Edge(2L, 3L, "登录"),
  Edge(3L, 1L, "担保")
))

// 顶点属性默认为 1 (空值占位)
val graphFromEdges: Graph[Int, String] = Graph.fromEdges(edgesOnly, defaultValue = 1)

// 后续通过 mapVertices 补充真实属性
val enrichedGraph = graphFromEdges.mapVertices { (id, _) =>
  // 从外部表查询用户风险分
  lookupRiskScore(id)
}

方式三：从 CSV/Parquet 文件加载

生产环境中，顶点和边通常存储在分布式文件系统中。推荐先加载为 DataFrame，再转换为 RDD 构建图。

// 1. 从 Parquet 加载顶点表
val vertexDF = spark.read.parquet("/data/fraud/vertices")
// 假设 schema: vertex_id: Long, risk_label: Int, device_id: String

// 2. 从 Parquet 加载边表
val edgeDF = spark.read.parquet("/data/fraud/edges")
// 假设 schema: src_id: Long, dst_id: Long, amount: Double, timestamp: Long

// 3. 转换为 RDD 并构建图
val vertexRDD: RDD[(VertexId, (Int, String))] = vertexDF.rdd.map(row => (
  row.getLong(0),
  (row.getInt(1), row.getString(2))
))

val edgeRDD: RDD[Edge[Double]] = edgeDF.rdd.map(row => Edge(
  row.getLong(0),
  row.getLong(1),
  row.getDouble(2)
))

val graph: Graph[(Int, String), Double] = Graph(vertexRDD, edgeRDD)

下表总结了三种创建方式的适用场景：

4.4 创建流程总览

下图展示了从原始数据到 PropertyGraph 的完整创建流程，以及后续图计算在欺诈评分中的位置：