加载数据集后如何在Apache Spark Java中创建GraphX图

你的实现流程存在多处逻辑错误与类型不匹配问题，是导致报错的核心原因，具体问题如下：

• 类型不匹配错误：
  1. 你创建的Edge实例泛型为Edge<Double>（存储边长Double类型），但强制类型转换为Collection<? extends Edge<Integer>>，泛型不兼容会直接抛出类转换异常
  2. 定义的Graph泛型为Graph<Integer, Integer>，但fromEdgeTuples的第二个参数顶点默认值传入了字符串" "，和顶点泛型Integer完全不匹配
• 分布式算子使用错误：
  你在分布式算子foreach中往Driver端的本地ArrayListedges添加数据，Spark算子运行在Executor端，对Driver端本地集合的修改不会被同步，最终edges要么是空集合，要么仅包含部分节点的结果，数据量较大时还会直接触发Driver端OOM
• 冗余代码问题：
  你对同一份文本文件做了两次重复解析，第一次用map得到edgeFileRDD，第二次用mapPartitions得到edgesPart，属于完全重复的逻辑，无任何意义
• 隐含参数错误：
  GraphX要求顶点ID必须为Long类型，你直接使用Integer类型的Source_ID、Destination_ID作为顶点ID，也会触发类型不匹配报错

你不需要将分布式RDD转为本地集合再转RDD，直接在RDD上做转换即可，修正后代码如下：

1. EdgeNetwork.java 无需修改，保持原实现即可 #

2. GraphDatasetParsingFile.java 修正代码 #

package graphXApi;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.graphx.Edge;
import org.apache.spark.graphx.Graph;
import org.apache.spark.storage.StorageLevel;
import scala.reflect.ClassTag;
import scala.reflect.ClassTag$;

public class GraphDatasetParsingFile {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local").setAppName("GraphFileReadClass");
        JavaSparkContext javaSparkContext = new JavaSparkContext(conf);
        // 定义顶点、边属性的ClassTag，和实际类型匹配
        ClassTag<String> vertexTag = ClassTag$.MODULE$.apply(String.class);
        ClassTag<Double> edgeTag = ClassTag$.MODULE$.apply(Double.class);

        JavaRDD<String> textFile = javaSparkContext.textFile("C:\\Users\\Aavash\\Desktop\\SFEdge.txt");

        // 仅解析一次文件得到EdgeNetwork RDD
        JavaRDD<EdgeNetwork> edgeFileRDD = textFile.map(line -> {
            String[] parts = line.split(" ");
            EdgeNetwork edgeNet = new EdgeNetwork();
            edgeNet.setEdge_id(Integer.parseInt(parts[0]));
            edgeNet.setSrc_id(Integer.parseInt(parts[1]));
            edgeNet.setDest_id(Integer.parseInt(parts[2]));
            edgeNet.setEdge_Length(Double.parseDouble(parts[3]));
            return edgeNet;
        });

        // 直接转换为GraphX要求的Edge RDD，无需拉取到Driver端
        JavaRDD<Edge<Double>> edgeRDD = edgeFileRDD.map(edgeNet -> 
            new Edge<>(edgeNet.getSrc_id().longValue(), edgeNet.getDest_id().longValue(), edgeNet.getEdge_Length())
        );

        // 构建图：顶点默认值设为空字符串，和vertexTag匹配，边属性为Double类型
        Graph<String, Double> graph = Graph.fromEdges(
                edgeRDD.rdd(),
                "",
                StorageLevel.MEMORY_ONLY(),
                StorageLevel.MEMORY_ONLY(),
                vertexTag,
                edgeTag
        );
        
        // 验证图构建结果
        System.out.println("图顶点总数：" + graph.vertices().count());
        System.out.println("图边总数：" + graph.edges().count());
    }
}

可选优化 #

如果你不需要保留Edge_id字段，整个EdgeNetwork类也可以省略，直接在解析文本的时候就转换为Edge实例，减少不必要的对象创建开销。

内容的提问来源于stack exchange，提问作者Aavash Bhandari