如何从包含顶点与边的文本文件创建Spark GraphX图？

问题1：ClassTag类型不匹配（直接触发报错的核心原因） #

你定义ClassTag的代码存在明显类型错误，且创建Graph时传入的泛型标签参数不匹配：

• 错误定义ClassTag<String> intTag = scala.reflect.ClassTag$.MODULE$.apply(Integer.class);，变量接收类型和实际生成的Integer类型ClassTag完全不符
• Graph<String, Double>的边属性泛型是Double，但你调用Graph.fromEdges时最后两个参数都传了stringTag，第二个标签应该对应边属性Double的ClassTag，类型对不上直接触发泛型校验报错。

问题2：RDD构建逻辑存在分布式执行常识性错误 #

你在Driver端定义了本地集合nodes和edges，然后在mapPartitions算子中往这两个集合里添加元素，是Spark开发的典型错误：

• Spark的转换算子（比如mapPartitions）是惰性执行的，且逻辑运行在Executor节点上，Driver端的本地集合根本无法收集到Executor上生成的数据
• 在你调用javaSparkContext.parallelize(nodes)和javaSparkContext.parallelize(edges)时，mapPartitions还没有执行，这两个本地集合是空的，就算后续触发了行动算子，分布式节点上的更新也不会同步回Driver的本地集合，最终生成的edgesRDD和nodesRDD都是空的或者数据不全。

修正代码示例 #

public static void main(String[] args) {
    SparkConf conf = new SparkConf().setMaster("local").setAppName("GraphFileReadClass");
    JavaSparkContext javaSparkContext = new JavaSparkContext(conf);
    // 正确定义对应类型的ClassTag
    ClassTag<String> vdTag = scala.reflect.ClassTag$.MODULE$.apply(String.class);
    ClassTag<Double> edTag = scala.reflect.ClassTag$.MODULE$.apply(Double.class);

    // 读取边文件直接生成Edge<Double>类型RDD
    JavaRDD<String> inputEdgesTextFile = javaSparkContext.textFile("./SFEdges.txt");
    JavaRDD<Edge<Double>> edgesRDD = inputEdgesTextFile.map(line -> {
        String[] parts = line.split(" ");
        return new Edge<Double>(
            Long.parseLong(parts[1]),
            Long.parseLong(parts[2]),
            Double.parseDouble(parts[3])
        );
    });

    // 直接生成Graph
    Graph<String, Double> graph = Graph.fromEdges(
        edgesRDD.rdd(),
        "默认顶点属性",
        StorageLevel.MEMORY_ONLY(),
        StorageLevel.MEMORY_ONLY(),
        vdTag,
        edTag
    );
}

如果你需要给顶点设置自定义的经纬度属性，不需要用默认值，可以直接构造顶点RDD调用Graph.apply方法创建图：

// 构造顶点RDD，格式是(顶点ID, 顶点属性)
JavaRDD<Tuple2<Object, NodesClass>> nodesRDD = inputNodesTextFile.map(line -> {
    String[] parts = line.split(" ");
    NodesClass node = new NodesClass();
    node.setNode_Id(Integer.parseInt(parts[0]));
    node.setLongitude(Double.parseDouble(parts[1]));
    node.setLatitude(Double.parseDouble(parts[2]));
    return new Tuple2<Object, NodesClass>(Long.parseLong(parts[0]), node);
});
// 对应修改顶点属性的ClassTag为NodesClass的类型
ClassTag<NodesClass> vdTag = scala.reflect.ClassTag$.MODULE$.apply(NodesClass.class);
Graph<NodesClass, Double> graph = Graph.apply(
    nodesRDD.rdd(),
    edgesRDD.rdd(),
    new NodesClass(), // 边中存在但顶点RDD中没有的顶点的默认属性
    StorageLevel.MEMORY_ONLY(),
    StorageLevel.MEMORY_ONLY(),
    vdTag,
    edTag
);

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