import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Test
public void testAccumulatorsAfterNoOp() {

	final String accName = "test_accumulator";

	try {
		env.setParallelism(6);
		env.getConfig().disableSysoutLogging();

		env.generateSequence(1, 1000000)
				.rebalance()
				.flatMap(new RichFlatMapFunction<Long, Long>() {

					private LongCounter counter;

					@Override
					public void open(Configuration parameters) {
						counter = getRuntimeContext().getLongCounter(accName);
					}

					@Override
					public void flatMap(Long value, Collector<Long> out) {
						counter.add(1L);
					}
				})
				.output(new DiscardingOutputFormat<Long>());

		JobExecutionResult result = env.execute();

		assertEquals(1000000L, result.getAllAccumulatorResults().get(accName));
	}
	catch (Exception e) {
		e.printStackTrace();
		fail(e.getMessage());
	}
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public void close() throws IOException {
	addAccumulator("vc", new LongCounter(vertexCount));
	addAccumulator("uec", new LongCounter(unidirectionalEdgeCount));
	addAccumulator("bec", new LongCounter(bidirectionalEdgeCount));
	addAccumulator("021d", new LongCounter(triplet021dCount));
	addAccumulator("021u", new LongCounter(triplet021uCount));
	addAccumulator("021c", new LongCounter(triplet021cCount));
	addAccumulator("111d", new LongCounter(triplet111dCount));
	addAccumulator("111u", new LongCounter(triplet111uCount));
	addAccumulator("201", new LongCounter(triplet201Count));
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public void close() throws IOException {
	addAccumulator(VERTEX_COUNT, new LongCounter(vertexCount));
	addAccumulator(UNIDIRECTIONAL_EDGE_COUNT, new LongCounter(unidirectionalEdgeCount));
	addAccumulator(BIDIRECTIONAL_EDGE_COUNT, new LongCounter(bidirectionalEdgeCount));
	addAccumulator(TRIPLET_COUNT, new LongCounter(tripletCount));
	addAccumulator(MAXIMUM_DEGREE, new LongMaximum(maximumDegree));
	addAccumulator(MAXIMUM_OUT_DEGREE, new LongMaximum(maximumOutDegree));
	addAccumulator(MAXIMUM_IN_DEGREE, new LongMaximum(maximumInDegree));
	addAccumulator(MAXIMUM_TRIPLETS, new LongMaximum(maximumTriplets));
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public void close() throws IOException {
	addAccumulator(TRIANGLE_TRIPLET_COUNT, new LongCounter(triangleTripletCount));
	addAccumulator(RECTANGLE_TRIPLET_COUNT, new LongCounter(rectangleTripletCount));
	addAccumulator(MAXIMUM_TRIANGLE_TRIPLETS, new LongMaximum(maximumTriangleTriplets));
	addAccumulator(MAXIMUM_RECTANGLE_TRIPLETS, new LongMaximum(maximumRectangleTriplets));
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public void close() throws IOException {
	addAccumulator(VERTEX_COUNT, new LongCounter(vertexCount));
	addAccumulator(EDGE_COUNT, new LongCounter(edgeCount));
	addAccumulator(TRIPLET_COUNT, new LongCounter(tripletCount));
	addAccumulator(MAXIMUM_DEGREE, new LongMaximum(maximumDegree));
	addAccumulator(MAXIMUM_TRIPLETS, new LongMaximum(maximumTriplets));
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public void increment(String group, String counter, long l) {
	if(this.runtimeContext != null) {
		LongCounter flinkCounter = getOrInitCounter(EnumStringConverter.mergeGroupCounter(group, counter));
		flinkCounter.add(l);
	}
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public LongCounter clone() {
	throw new CustomException();
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public void open(Configuration parameters) throws Exception {
	getRuntimeContext().addAccumulator("test", new LongCounter());
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public LongCounter getLongCounter(String name) {
	return (LongCounter) getAccumulator(name, LongCounter.class);
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public void close() throws IOException {
	addAccumulator(COUNT, new LongCounter(count));
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public void close() throws IOException {
	addAccumulator(VERTEX_COUNT, new LongCounter(vertexCount));
	addAccumulator(SUM_OF_LOCAL_CLUSTERING_COEFFICIENT, new DoubleCounter(sumOfLocalClusteringCoefficient));
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Test
public void testHandleRequest() throws Exception {

	// Instance the handler.
	final RestHandlerConfiguration restHandlerConfiguration = RestHandlerConfiguration.fromConfiguration(new Configuration());

	final SubtaskExecutionAttemptAccumulatorsHandler handler = new SubtaskExecutionAttemptAccumulatorsHandler(
		CompletableFuture.completedFuture("127.0.0.1:9527"),
		() -> null,
		Time.milliseconds(100L),
		restHandlerConfiguration.getResponseHeaders(),
		null,
		new ExecutionGraphCache(
			restHandlerConfiguration.getTimeout(),
			Time.milliseconds(restHandlerConfiguration.getRefreshInterval())),
		TestingUtils.defaultExecutor());

	// Instance a empty request.
	final HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request = new HandlerRequest<>(
		EmptyRequestBody.getInstance(),
		new SubtaskAttemptMessageParameters()
	);

	final Map<String, Accumulator<?, ?>> userAccumulators = new HashMap<>(2);
	userAccumulators.put("IntCounter", new IntCounter(10));
	userAccumulators.put("LongCounter", new LongCounter(100L));

	// Instance the expected result.
	final StringifiedAccumulatorResult[] accumulatorResults =
		StringifiedAccumulatorResult.stringifyAccumulatorResults(userAccumulators);

	final int attemptNum = 1;
	final int subtaskIndex = 2;

	// Instance the tested execution.
	final ArchivedExecution execution = new ArchivedExecution(
		accumulatorResults,
		null,
		new ExecutionAttemptID(),
		attemptNum,
		ExecutionState.FINISHED,
		null,
		null,
		subtaskIndex,
		new long[ExecutionState.values().length]);

	// Invoke tested method.
	final SubtaskExecutionAttemptAccumulatorsInfo accumulatorsInfo = handler.handleRequest(request, execution);

	final ArrayList<UserAccumulator> userAccumulatorList = new ArrayList<>(userAccumulators.size());
	for (StringifiedAccumulatorResult accumulatorResult : accumulatorResults) {
		userAccumulatorList.add(
			new UserAccumulator(
				accumulatorResult.getName(),
				accumulatorResult.getType(),
				accumulatorResult.getValue()));
	}

	final SubtaskExecutionAttemptAccumulatorsInfo expected = new SubtaskExecutionAttemptAccumulatorsInfo(
		subtaskIndex,
		attemptNum,
		execution.getAttemptId().toString(),
		userAccumulatorList);

	// Verify.
	assertEquals(expected, accumulatorsInfo);
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public LongCounter getLongCounter(String name) {
	throw new UnsupportedOperationException("Long counters are not supported in rich async functions.");
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public LongCounter getLongCounter(String name) {
	throw new UnsupportedOperationException();
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
public static void main(final String[] args) throws Exception {
	// set up the execution environment
	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
	String in = args[0];
	String out = args[1];
	System.err.println("Using input=" + in);
	System.err.println("Using output=" + out);

	String patterns[] = new String[args.length - 2];
	System.arraycopy(args, 2, patterns, 0, args.length - 2);
	System.err.println("Using patterns: " + Arrays.toString(patterns));

	// get input data
	DataSet<String> text = env.readTextFile(args[0]);
	for (int p = 0; p < patterns.length; p++) {
		final String pattern = patterns[p];
		DataSet<String> res = text.filter(new RichFilterFunction<String>() {
			private static final long serialVersionUID = 1L;

			Pattern p = Pattern.compile(pattern);
			LongCounter filterMatches = new LongCounter();
			LongCounter filterRecords = new LongCounter();

			@Override
			public void open(Configuration parameters) throws Exception {
				super.open(parameters);
				getRuntimeContext().addAccumulator("filterMatchCount-" + pattern, filterMatches);
				getRuntimeContext().addAccumulator("filterRecordCount-" + pattern, filterRecords);
			}

			@Override
			public boolean filter(String value) throws Exception {
				filterRecords.add(1L);
				if (value == null || value.length() == 0) {
					return false;
				}
				final Matcher m = p.matcher(value);
				if (m.find()) {
					filterMatches.add(1L);
					return true;
				}
				return false;
			}
		}).name("grep for " + pattern);
		res.writeAsText(out + "_" + pattern, FileSystem.WriteMode.OVERWRITE);
	}

	// execute program
	JobExecutionResult jobResult = env.execute("Flink Grep benchmark");
	System.err.println(AccumulatorHelper.getResultsFormated(jobResult.getAllAccumulatorResults()));
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
@Override
public LongCounter getLongCounter(String name) {
	return context.getLongCounter(name);
}
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
/**
 * Convenience function to create a counter object for longs.
 */
@PublicEvolving
LongCounter getLongCounter(String name);
 

import org.apache.flink.api.common.accumulators.LongCounter; //导入依赖的package包/类
/**
 * Convenience function to create a counter object for longs.
 */
LongCounter getLongCounter(String name);