【Flink图计算源码解析】：Flink图的存储数据结构

承接上篇：https://blog.csdn.net/hxcaifly/article/details/86147736

1.图的基本组成

图由边和顶点构成：

• Edge: 边。每条边是Tuple3<K,K,V>的数据结构，保存了边的开始顶点Id，边的目的顶点Id和边的值。
• Vertex: 顶点。每个顶点是Tuple2<K, V>的数据结构。保存了顶点的Id，和顶点的值。

2. 顶点和边的代码定义

2.1. 顶点的定义

package org.apache.flink.graph;

import org.apache.flink.api.java.tuple.Tuple2;

/**
 * 图的顶点。由ID和值构成
 * 对于没有值的顶点，利用 {@link org.apache.flink.types.NullValue} 作为值类型。
 *
 * @param <K>
 * @param <V>
 */
public class Vertex<K, V> extends Tuple2<K, V> {
   

	private static final long serialVersionUID = 1L;

	public Vertex(){
   }

	public Vertex(K k, V val) {
   
		this.f0 = k;
		this.f1 = val;
	}

	public K getId() {
   
		return this.f0;
	}

	public V getValue() {
   
		return this.f1;
	}

	public void setId(K id) {
   
		this.f0 = id;
	}

	public void setValue(V val) {
   
		this.f1 = val;
	}
}

顶点的定义其实很简洁，顶点就是由顶点Id和顶点值构成。所以就利用了Tuple2<K, V> 数据结构。其中K表示顶点Id的类型；V表示顶点的值类型。

2.2. 边的定义

package org.apache.flink.graph;

import org.apache.flink.api.java.tuple.Tuple3;

/**
 * 边是呈现两个顶点{@link Vertex vertices}之间的连线
 * 对于边如果没有值，用{@link org.apache.flink.types.NullValue}作为值类型。
 *
 * @param <K> 源顶点和目的顶点的key类型
 * @param <V> 边的值类型
 */
public class Edge<K, V> extends Tuple3<K, K, V> {
   

	private static final long serialVersionUID = 1L;

	public Edge(){
   }

	public Edge(K source, K target, V value) {
   
		this.f0 = source;
		this.f1 = target;
		this.f2 = value;
	}

	/**
	 * Reverses the direction of this Edge.
	 * @return a new Edge, where the source is the original Edge's target
	 * and the target is the original Edge's source.
	 */
	public Edge<K, V> reverse() {
   
			return new Edge<>(this.f1, this.f0, this.f2);
	}

	public void setSource(K source) {
   
		this.f0 = source;
	}

	public K getSource() {
   
		return this.f0;
	}

	public void setTarget(K target) {
   
		this.f1 = target;
	}

	public K getTarget() {
   
		return f1;
	}

	public void setValue(V value) {
   
		this.f2 = value;
	}

	public V getValue() {
   
		return f2;
	}
}

边是表示两个顶点之间的连线。那么怎样才能记录一条边呢。很容易想到，需要记录边的出发点（起始顶点）和目的点（目标顶点）。然后两个顶点之间的连线是有值的，这个值可以表示不同的含义，比如可以表示权重，或者表示距离。

所以，边的数据结构是继承了Tuple3<K, K, V>。其中，K表示顶点的类型，分别是起始顶点和目标顶点；V表示边的值类型。

2.3. 图的定义

我们已经知道图（Graph）是由边和顶点构成的，那么我们不难想象Graph这个类的定义。

package org.apache.flink.graph;

/**
 * 定义由边{@link Edge edges}和顶点{@link Vertex vertices}组成的图。
 *  * @see org.apache.flink.graph.Edge
 * @see org.apache.flink.graph.Vertex
 *  *  * @param <K>  顶点的key类型
 * @param <VV> 顶点的值类型
 * @param <EV> 边的值类型
 */
@SuppressWarnings("serial")
public class Graph<K, VV, EV> {
   
    
    // 1. 执行环境
	private final ExecutionEnvironment context;
    // 2.图的顶点数据集
	private final DataSet<Vertex<K, VV>> vertices;
    
    // 3. 图的边数据集
	private final DataSet<Edge<K, EV>> edges;

    // 省略了方法操作等
}

上述代码是图的定义中的成员变量部分。

• ExecutionEnvironment ：图的执行上下文，图的计算其实也是一个单独的任务。需要依靠于这个执行上下文去获取一些相关执行配置等。
• DataSet<Vertex<K, VV>> ： 顶点的数据集。
• DataSet<Edge<K, EV>>： 边的数据集。

Graph类中定义了很多构造方法，这些构造方法主要是要适应从不同类型的数据源中读取图数据集。比如从csv文件中读取，或者说只提供了边数据集时，我们需要从顶点数据中提取出所有的顶点。这一块可能会随着业务的变化会有所拓展，因为宗旨就是方便不同的开发者能够多样化读入数据源。
下面列出一部分构建函数：

/**
	 * 从两个DataSets中创建图：顶点和边
	 *
	 * @param vertices 顶点的DataSet.
	 * @param edges 边的DataSet.
	 * @param context flink执行环境.
	 */
	protected Graph(DataSet<Vertex<K, VV>> vertices, DataSet<Edge<K, EV>> edges, ExecutionEnvironment context) {
   
		this.vertices = vertices