基于Spark DataFrame创建带属性GraphX时解决ClassCastException

你遇到的ClassCastException完全来自边构造时的类型不匹配，和PageRank、最短路径算法本身无关：

• 边表中weight字段实际是Double类型（从样例输出可以看到值带小数点，对应DataFrame schema里的DoubleType），但你构造Edge[Long]时强行用getAs[Long]读取该字段，执行阶段反序列化时就会触发类型转换错误。
• 想要保留所有边业务字段的需求完全可以实现，不需要把属性硬塞成裸元组，用自定义Case Class承载即可，可读性和类型安全性都会高很多。

1. 修正RDD构造逻辑，完全对齐DataFrame schema #

先定义两个Case Class分别承载节点、边的业务属性，避免裸元组的类型推断问题：

// 节点属性：匹配vertices_complete的全部业务字段
case class VertexAttr(
  node: String,
  trip_id: Long,
  stop_id: String,
  time: Long
)

// 边属性：保留你需要的全部业务字段
case class EdgeAttr(
  time: Long,
  method: String,
  weight: Double, // 严格对齐DataFrame中weight的Double类型
  walk_time: Long,
  wait_time: Long,
  bus_time: Long
)

基于Case Class生成符合GraphX要求的RDD结构：

// 节点RDD：严格符合(VertexId, 节点属性)的格式要求
val verticesRDD: RDD[(VertexId, VertexAttr)] = vertices_complete.rdd.map(row => {
  (
    row.getAs[Long]("id"),
    VertexAttr(
      node = row.getAs[String]("node"),
      trip_id = row.getAs[Long]("trip_id"),
      stop_id = row.getAs[String]("stop_id"),
      time = row.getAs[Long]("time")
    )
  )
})

// 边RDD：严格符合Edge[边属性]的格式要求
val edgesRDD: RDD[Edge[EdgeAttr]] = edges_DF.rdd.map(row => {
  Edge(
    srcId = row.getAs[Long]("src"),
    dstId = row.getAs[Long]("dst"),
    attr = EdgeAttr(
      time = row.getAs[Long]("time"),
      method = row.getAs[String]("method"),
      weight = row.getAs[Double]("weight"),
      walk_time = row.getAs[Long]("walk_time"),
      wait_time = row.getAs[Long]("wait_time"),
      bus_time = row.getAs[Long]("bus_time")
    )
  )
})

基于修正后的RDD生成Graph对象：

val my_graph: Graph[VertexAttr, EdgeAttr] = Graph(verticesRDD, edgesRDD)

2. 图分析任务的正常调用 #

GraphX内置算法会自动适配自定义的边属性类型，不会因为存储了额外业务字段触发错误，示例调用如下：

// PageRank计算
val ranks = my_graph.pageRank(0.0001).vertices

// 基于边权重的单源最短路径计算
val shortestPaths = my_graph.ops.pregel[Double](
  initialMsg = Double.PositiveInfinity,
  maxIterations = 10
)(
  vprog = (_, dist, msg) => math.min(dist, msg),
  sendMsg = (triplet) => {
    if (triplet.srcAttr != Double.PositiveInfinity) {
      Iterator((triplet.dstId, triplet.srcAttr + triplet.attr.weight))
    } else Iterator.empty
  },
  mergeMsg = (a, b) => math.min(a, b)
)

内容的提问来源于stack exchange，提问作者Andrea Nicolai