import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Test
public void testUnion() throws Exception {
	/*
	 * Test union()
	 */
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

	Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
		TestGraphUtils.getLongLongEdgeData(env), env);

	List<Vertex<Long, Long>> vertices = new ArrayList<>();
	List<Edge<Long, Long>> edges = new ArrayList<>();

	vertices.add(new Vertex<>(6L, 6L));
	edges.add(new Edge<>(6L, 1L, 61L));

	graph = graph.union(Graph.fromCollection(vertices, edges, env));

	DataSet<Edge<Long, Long>> data = graph.getEdges();
	List<Edge<Long, Long>> result = data.collect();

	expectedResult = "1,2,12\n" +
		"1,3,13\n" +
		"2,3,23\n" +
		"3,4,34\n" +
		"3,5,35\n" +
		"4,5,45\n" +
		"5,1,51\n" +
		"6,1,61\n";

	compareResultAsTuples(result, expectedResult);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Test
public void testFromTuple2WithMapper() throws Exception {
	/*
	 * Test graph creation with fromTuple2DataSet with vertex initializer
	 */
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
	DataSet<Tuple2<Long, Long>> edges = TestGraphUtils.getLongLongTuple2Data(env);

	Graph<Long, String, NullValue> graph = Graph.fromTuple2DataSet(edges,
		new BooMapper(), env);

	List<Vertex<Long, String>> result = graph.getVertices().collect();

	expectedResult = "1,boo\n" +
		"2,boo\n" +
		"3,boo\n" +
		"4,boo\n" +
		"6,boo\n" +
		"10,boo\n" +
		"20,boo\n" +
		"30,boo\n" +
		"40,boo\n" +
		"60,boo\n";

	compareResultAsTuples(result, expectedResult);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Override
public void compute(Vertex<K, Double> vertex, MessageIterator<NewMinDistance> messageIterator) throws Exception {
    // Send initial group of messages from the source vertex
    if (vertex.getId().equals(sourceVertex) && getSuperstepNumber() == 1) {
        sendNewDistanceToAll(0);
    }

    // Calculate new min distance from source node
    double minDistance = minDistance(messageIterator);

    // Send new min distance to neighbour vertices if new min distance is less
    if (minDistance < vertex.getValue()) {
        setNewVertexValue(minDistance);
        sendNewDistanceToAll(minDistance);
    }
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Test
public void testValidate() throws Exception {
	/*
	 * Test validate():
	 */
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
	DataSet<Vertex<Long, Long>> vertices = TestGraphUtils.getLongLongVertexData(env);
	DataSet<Edge<Long, Long>> edges = TestGraphUtils.getLongLongEdgeData(env);

	Graph<Long, Long, Long> graph = Graph.fromDataSet(vertices, edges, env);
	Boolean valid = graph.validate(new InvalidVertexIdsValidator<>());

	//env.fromElements(result).writeAsText(resultPath);

	String res = valid.toString(); //env.fromElements(valid);
	List<String> result = new LinkedList<>();
	result.add(res);
	expectedResult = "true";

	compareResultAsText(result, expectedResult);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Test
public void testWithCustomParametrizedType() throws Exception {
	/*
	 * Test mapVertices() and change the value type to a parameterized custom type
	 */
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

	Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
		TestGraphUtils.getLongLongEdgeData(env), env);

	DataSet<Vertex<Long, DummyCustomParameterizedType<Double>>> mappedVertices = graph.mapVertices(
		new ToCustomParametrizedTypeMapper()).getVertices();
	List<Vertex<Long, DummyCustomParameterizedType<Double>>> result = mappedVertices.collect();

	expectedResult = "1,(1.0,1)\n" +
		"2,(2.0,2)\n" +
		"3,(3.0,3)\n" +
		"4,(4.0,4)\n" +
		"5,(5.0,5)\n";

	compareResultAsTuples(result, expectedResult);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Test
public void testAddVertex() throws Exception {
	/*
	 * Test addVertex() -- simple case
	 */

	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

	Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
			TestGraphUtils.getLongLongEdgeData(env), env);

	graph = graph.addVertex(new Vertex<>(6L, 6L));

	DataSet<Vertex<Long, Long>> data = graph.getVertices();
	List<Vertex<Long, Long>> result = data.collect();

	expectedResult = "1,1\n" +
			"2,2\n" +
			"3,3\n" +
			"4,4\n" +
			"5,5\n" +
			"6,6\n";

	compareResultAsTuples(result, expectedResult);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Test
public void testRemoveOneValidOneInvalidVertex() throws Exception {
	/*
	 * Test removeVertices() -- remove one invalid vertex and a valid one
	 */

	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

	Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
			TestGraphUtils.getLongLongEdgeData(env), env);
	List<Vertex<Long, Long>> verticesToBeRemoved = new ArrayList<>();
	verticesToBeRemoved.add(new Vertex<>(1L, 1L));
	verticesToBeRemoved.add(new Vertex<>(7L, 7L));

	graph = graph.removeVertices(verticesToBeRemoved);

	DataSet<Edge<Long, Long>> data = graph.getEdges();
	List<Edge<Long, Long>> result = data.collect();

	expectedResult = "2,3,23\n" +
			"3,4,34\n" +
			"3,5,35\n" +
			"4,5,45\n";

	compareResultAsTuples(result, expectedResult);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Test
public void testWithStringValue() throws Exception {
	/*
	 * Test mapVertices() and change the value type to String
	 */
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

	Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
		TestGraphUtils.getLongLongEdgeData(env), env);

	DataSet<Vertex<Long, String>> mappedVertices = graph.mapVertices(new ToStringMapper()).getVertices();
	List<Vertex<Long, String>> result = mappedVertices.collect();

	expectedResult = "1,one\n" +
		"2,two\n" +
		"3,three\n" +
		"4,four\n" +
		"5,five\n";

	compareResultAsTuples(result, expectedResult);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Override
public DataSet<Vertex<K, Tuple2<DoubleValue, DoubleValue>>> run(Graph<K, VV, EV> graph) throws Exception {
	Graph<K, Tuple2<DoubleValue, DoubleValue>, Boolean> newGraph = graph
			.mapEdges(new AuthorityEdgeMapper<K, EV>())
			.union(graph.reverse().mapEdges(new HubEdgeMapper<K, EV>()))
			.mapVertices(new VertexInitMapper<K, VV>());

	ScatterGatherConfiguration parameter = new ScatterGatherConfiguration();
	parameter.setDirection(EdgeDirection.OUT);
	parameter.setOptNumVertices(true);
	parameter.registerAggregator("updatedValueSum", new DoubleSumAggregator());
	parameter.registerAggregator("authorityValueSum", new DoubleSumAggregator());
	parameter.registerAggregator("diffValueSum", new DoubleSumAggregator());

	return newGraph
			.runScatterGatherIteration(new MessageUpdate<K>(maxIterations),
					new VertexUpdate<K>(maxIterations, convergeThreshold), maxIterations, parameter)
			.getVertices();
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Override
public void sendMessages(Vertex<K, Tuple2<DoubleValue, DoubleValue>> vertex) {
	// in the first iteration, no aggregation to call, init sum with value of vertex
	double iterationValueSum = 1.0;

	if (getSuperstepNumber() > 1) {
		iterationValueSum = Math.sqrt(((DoubleValue) getPreviousIterationAggregate("updatedValueSum")).getValue());
	}
	for (Edge<K, Boolean> edge : getEdges()) {
		if (getSuperstepNumber() != maxIteration) {
			if (getSuperstepNumber() % 2 == 1) {
				if (edge.getValue()) {
					sendMessageTo(edge.getTarget(), vertex.getValue().f0.getValue() / iterationValueSum);
				}
			} else {
				if (!edge.getValue()) {
					sendMessageTo(edge.getTarget(), vertex.getValue().f1.getValue() / iterationValueSum);
				}
			}
		} else {
			if (!edge.getValue()) {
				sendMessageTo(edge.getTarget(), iterationValueSum);
			}
		}
	}
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Test
public void testConnectedComponentsWithObjectReuseEnabled() throws Exception {
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
	env.getConfig().enableObjectReuse();

	DataSet<Edge<LongValue, NullValue>> edges = Translate.translateEdgeIds(
		ConnectedComponentsDefaultData.getDefaultEdgeDataSet(env),
		new LongToLongValue());

	Graph<LongValue, LongValue, NullValue> inputGraph = Graph.fromDataSet(
		edges, new IdentityMapper<>(), env);

	List<Vertex<LongValue, LongValue>> result = inputGraph.run(
		new GSAConnectedComponents<>(16)).collect();

	compareResultAsTuples(result, expectedResultCC);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
public String map(Vertex<Long, Long> vertex) throws Exception {
	String stringValue;
	if (vertex.getValue() == 1) {
		stringValue = "one";
	} else if (vertex.getValue() == 2) {
		stringValue = "two";
	} else if (vertex.getValue() == 3) {
		stringValue = "three";
	} else if (vertex.getValue() == 4) {
		stringValue = "four";
	} else if (vertex.getValue() == 5) {
		stringValue = "five";
	} else {
		stringValue = "";
	}
	return stringValue;
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Override
public void writeRecord(Vertex<T, Degrees> record) throws IOException {
	long degree = record.f1.getDegree().getValue();
	long outDegree = record.f1.getOutDegree().getValue();
	long inDegree = record.f1.getInDegree().getValue();

	long bidirectionalEdges = outDegree + inDegree - degree;
	long triplets = degree * (degree - 1) / 2;

	vertexCount++;
	unidirectionalEdgeCount += degree - bidirectionalEdges;
	bidirectionalEdgeCount += bidirectionalEdges;
	tripletCount += triplets;
	maximumDegree = Math.max(maximumDegree, degree);
	maximumOutDegree = Math.max(maximumOutDegree, outDegree);
	maximumInDegree = Math.max(maximumInDegree, inDegree);
	maximumTriplets = Math.max(maximumTriplets, triplets);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Test
public void testSingleIteration() throws Exception {
	/*
	 * Test one iteration of the Simple Community Detection Example
	 */

	ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
	Graph<Long, Long, Double> inputGraph = Graph.fromDataSet(
		CommunityDetectionData.getSimpleEdgeDataSet(env), new InitLabels(), env);

	List<Vertex<Long, Long>> result = inputGraph.run(new CommunityDetection<>(1, CommunityDetectionData.DELTA))
		.getVertices().collect();

	expected = CommunityDetectionData.COMMUNITIES_SINGLE_ITERATION;
	compareResultAsTuples(result, expected);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Test
public void testWithEmptyGraphWithVertices() throws Exception {
	DataSet<Vertex<LongValue, LongValue>> inDegreeWithoutZeroDegreeVertices = emptyGraphWithVertices
		.run(new VertexInDegree<LongValue, NullValue, NullValue>()
			.setIncludeZeroDegreeVertices(false));

	assertEquals(0, inDegreeWithoutZeroDegreeVertices.collect().size());

	DataSet<Vertex<LongValue, LongValue>> inDegreeWithZeroDegreeVertices = emptyGraphWithVertices
		.run(new VertexInDegree<LongValue, NullValue, NullValue>()
			.setIncludeZeroDegreeVertices(true));

	String expectedResult =
		"(0,0)\n" +
		"(1,0)\n" +
		"(2,0)";

	TestBaseUtils.compareResultAsText(inDegreeWithZeroDegreeVertices.collect(), expectedResult);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
private <VV extends Comparable<VV>> void validateVertices(String[] expectedVertices, List<Vertex<Long, Summarization.VertexValue<VV>>> actualVertices) {
	Arrays.sort(expectedVertices);
	Collections.sort(actualVertices, new Comparator<Vertex<Long, Summarization.VertexValue<VV>>>() {
		@Override
		public int compare(Vertex<Long, Summarization.VertexValue<VV>> o1, Vertex<Long, Summarization.VertexValue<VV>> o2) {
			int result = o1.getId().compareTo(o2.getId());
			if (result == 0) {
				result = o1.getValue().getVertexGroupValue().compareTo(o2.getValue().getVertexGroupValue());
			}
			return result;
		}
	});

	for (int i = 0; i < expectedVertices.length; i++) {
		validateVertex(expectedVertices[i], actualVertices.get(i));
	}
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Override
public Graph<LongValue, NullValue, NullValue> generate() {
	Preconditions.checkState(!dimensions.isEmpty(), "No dimensions added to GridGraph");

	// Vertices
	DataSet<Vertex<LongValue, NullValue>> vertices = GraphGeneratorUtils.vertexSequence(env, parallelism, vertexCount);

	// Edges
	LongValueSequenceIterator iterator = new LongValueSequenceIterator(0, this.vertexCount - 1);

	DataSet<Edge<LongValue, NullValue>> edges = env
		.fromParallelCollection(iterator, LongValue.class)
			.setParallelism(parallelism)
			.name("Edge iterators")
		.flatMap(new LinkVertexToNeighbors(vertexCount, dimensions))
			.setParallelism(parallelism)
			.name("Grid graph edges");

	// Graph
	return Graph.fromDataSet(vertices, edges, env);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Override
public Graph<LongValue, NullValue, NullValue> generate() {
	Preconditions.checkState(vertexCount >= 2);

	// Vertices
	DataSet<Vertex<LongValue, NullValue>> vertices = GraphGeneratorUtils.vertexSequence(env, parallelism, vertexCount);

	// Edges
	LongValueSequenceIterator iterator = new LongValueSequenceIterator(1, this.vertexCount - 1);

	DataSet<Edge<LongValue, NullValue>> edges = env
		.fromParallelCollection(iterator, LongValue.class)
			.setParallelism(parallelism)
			.name("Edge iterators")
		.flatMap(new LinkVertexToCenter())
			.setParallelism(parallelism)
			.name("Star graph edges");

	// Graph
	return Graph.fromDataSet(vertices, edges, env);
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Override
public DataSet<Edge<K, Tuple2<EV, LongValue>>> runInternal(Graph<K, VV, EV> input)
		throws Exception {
	// s, d(s)
	DataSet<Vertex<K, LongValue>> vertexDegrees = input
		.run(new VertexDegree<K, VV, EV>()
			.setReduceOnTargetId(reduceOnTargetId.get())
			.setParallelism(parallelism));

	// s, t, d(s)
	return input.getEdges()
		.join(vertexDegrees, JoinHint.REPARTITION_HASH_SECOND)
		.where(0)
		.equalTo(0)
		.with(new JoinEdgeWithVertexDegree<>())
			.setParallelism(parallelism)
			.name("Edge source degree");
}
 

import org.apache.flink.graph.Vertex; //导入依赖的package包/类
@Test
public void testWithEmptyGraphWithVertices() throws Exception {
	DataSet<Vertex<LongValue, LongValue>> degree;

	degree = emptyGraphWithVertices
		.run(new VertexDegree<LongValue, NullValue, NullValue>()
			.setIncludeZeroDegreeVertices(false));

	assertEquals(0, degree.collect().size());

	degree = emptyGraphWithVertices
		.run(new VertexDegree<LongValue, NullValue, NullValue>()
			.setIncludeZeroDegreeVertices(true));

	String expectedResult =
		"(0,0)\n" +
		"(1,0)\n" +
		"(2,0)";

	TestBaseUtils.compareResultAsText(degree.collect(), expectedResult);
}