import org.apache.flink.api.common.functions.RichFilterFunction; //导入依赖的package包/类
@Test
public void testDistributedCacheWithIterations() throws Exception{
	File tempFile = new File(testPath);
	try (FileWriter writer = new FileWriter(tempFile)) {
		writer.write(testString);
	}

	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
	env.registerCachedFile(resultPath, testName);

	IterativeDataSet<Long> solution = env.fromElements(1L).iterate(2);
	solution.closeWith(env.generateSequence(1, 2).filter(new RichFilterFunction<Long>() {
		@Override
		public void open(Configuration parameters) throws Exception{
			File file = getRuntimeContext().getDistributedCache().getFile(testName);
			BufferedReader reader = new BufferedReader(new FileReader(file));
			String output = reader.readLine();
			reader.close();
			assertEquals(output, testString);
		}

		@Override
		public boolean filter(Long value) throws Exception {
			return false;
		}
	}).withBroadcastSet(solution, "SOLUTION")).output(new DiscardingOutputFormat<Long>());
	env.execute();
	expected = testString; // this will be a useless verification now.
}
 

import org.apache.flink.api.common.functions.RichFilterFunction; //导入依赖的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<StringValue> text = env.createInput(new TextValueInputFormat(new Path(in)));

	for (int p = 0; p < patterns.length; p++) {
		final String pattern = patterns[p];
		DataSet<StringValue> res = text.filter(new RichFilterFunction<StringValue>() {
			Pattern p = Pattern.compile(pattern);

			@Override
			public void open(Configuration parameters) throws Exception {
				super.open(parameters);
			}

			@Override
			public boolean filter(StringValue valueIn) throws Exception {
				final String value = valueIn.getValue();
				if (value == null || value.length() == 0) {
					return false;
				}
				final Matcher m = p.matcher(value);
				if (m.find()) {
					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.functions.RichFilterFunction; //导入依赖的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()));
}