Java RichGroupReduceFunction类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Java›Java编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Java中org.apache.flink.api.common.functions.RichGroupReduceFunction类的典型用法代码示例。如果您正苦于以下问题：Java RichGroupReduceFunction类的具体用法？Java RichGroupReduceFunction怎么用？Java RichGroupReduceFunction使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

RichGroupReduceFunction类属于org.apache.flink.api.common.functions包，在下文中一共展示了RichGroupReduceFunction类的15个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Java代码示例。

示例1: testBasicType

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void testBasicType() {
	// use getGroupReduceReturnTypes()
	RichGroupReduceFunction<?, ?> function = new RichGroupReduceFunction<Boolean, Boolean>() {
		private static final long serialVersionUID = 1L;

		@Override
		public void reduce(Iterable<Boolean> values, Collector<Boolean> out) throws Exception {
			// nothing to do
		}
	};

	TypeInformation<?> ti = TypeExtractor.getGroupReduceReturnTypes(function, (TypeInformation) TypeInfoParser.parse("Boolean"));

	Assert.assertTrue(ti.isBasicType());
	Assert.assertEquals(BasicTypeInfo.BOOLEAN_TYPE_INFO, ti);
	Assert.assertEquals(Boolean.class, ti.getTypeClass());

	// use getForClass()
	Assert.assertTrue(TypeExtractor.getForClass(Boolean.class).isBasicType());
	Assert.assertEquals(ti, TypeExtractor.getForClass(Boolean.class));

	// use getForObject()
	Assert.assertEquals(BasicTypeInfo.BOOLEAN_TYPE_INFO, TypeExtractor.getForObject(Boolean.valueOf(true)));
}

开发者ID:citlab，项目名称:vs.msc.ws14，代码行数:27，代码来源:TypeExtractorTest.java

示例2: testBasicType

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void testBasicType() {
	// use getGroupReduceReturnTypes()
	RichGroupReduceFunction<?, ?> function = new RichGroupReduceFunction<Boolean, Boolean>() {
		private static final long serialVersionUID = 1L;

		@Override
		public void reduce(Iterable<Boolean> values, Collector<Boolean> out) throws Exception {
			// nothing to do
		}
	};

	TypeInformation<?> ti = TypeExtractor.getGroupReduceReturnTypes(function, (TypeInformation) TypeInfoParser.parse("Boolean"));

	Assert.assertTrue(ti.isBasicType());
	Assert.assertEquals(BasicTypeInfo.BOOLEAN_TYPE_INFO, ti);
	Assert.assertEquals(Boolean.class, ti.getTypeClass());

	// use getForClass()
	Assert.assertTrue(TypeExtractor.getForClass(Boolean.class).isBasicType());
	Assert.assertEquals(ti, TypeExtractor.getForClass(Boolean.class));

	// use getForObject()
	Assert.assertEquals(BasicTypeInfo.BOOLEAN_TYPE_INFO, TypeExtractor.getForObject(true));
}

开发者ID:axbaretto，项目名称:flink，代码行数:27，代码来源:TypeExtractorTest.java

示例3: PlanUnwrappingReduceGroupOperator

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
public PlanUnwrappingReduceGroupOperator(GroupReduceFunction<IN, OUT> udf, Keys.SelectorFunctionKeys<IN, K> key, String name,
		TypeInformation<OUT> outType, TypeInformation<Tuple2<K, IN>> typeInfoWithKey, boolean combinable)
{
	super(combinable ? new TupleUnwrappingFlatCombinableGroupReducer<IN, OUT, K>((RichGroupReduceFunction<IN, OUT>) udf) : new TupleUnwrappingNonCombinableGroupReducer<IN, OUT, K>(udf),
			new UnaryOperatorInformation<Tuple2<K, IN>, OUT>(typeInfoWithKey, outType), key.computeLogicalKeyPositions(), name);
	
	super.setCombinable(combinable);
}

开发者ID:citlab，项目名称:vs.msc.ws14，代码行数:9，代码来源:PlanUnwrappingReduceGroupOperator.java

示例4: testAllGroupReduceNoCombiner

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
@Test
public void testAllGroupReduceNoCombiner() {
	try {
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(8);
		
		DataSet<Double> data = env.fromElements(0.2, 0.3, 0.4, 0.5).name("source");
		
		data.reduceGroup(new RichGroupReduceFunction<Double, Double>() {
			public void reduce(Iterable<Double> values, Collector<Double> out) {}
		}).name("reducer")
		.output(new DiscardingOutputFormat<Double>()).name("sink");
		
		Plan p = env.createProgramPlan();
		OptimizedPlan op = compileNoStats(p);
		
		OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
		
		
		// the all-reduce has no combiner, when the parallelism of the input is one
		
		SourcePlanNode sourceNode = resolver.getNode("source");
		SingleInputPlanNode reduceNode = resolver.getNode("reducer");
		SinkPlanNode sinkNode = resolver.getNode("sink");
		
		// check wiring
		assertEquals(sourceNode, reduceNode.getInput().getSource());
		assertEquals(reduceNode, sinkNode.getInput().getSource());
		
		// check that reduce has the right strategy
		assertEquals(DriverStrategy.ALL_GROUP_REDUCE, reduceNode.getDriverStrategy());
		
		// check parallelism
		assertEquals(1, sourceNode.getParallelism());
		assertEquals(1, reduceNode.getParallelism());
		assertEquals(1, sinkNode.getParallelism());
	}
	catch (Exception e) {
		System.err.println(e.getMessage());
		e.printStackTrace();
		fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
	}
}

开发者ID:axbaretto，项目名称:flink，代码行数:44，代码来源:GroupReduceCompilationTest.java

示例5: testGroupedReduceWithFieldPositionKeyNonCombinable

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
@Test
public void testGroupedReduceWithFieldPositionKeyNonCombinable() {
	try {
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(8);
		
		DataSet<Tuple2<String, Double>> data = env.readCsvFile("file:///will/never/be/read").types(String.class, Double.class)
			.name("source").setParallelism(6);
		
		data
			.groupBy(1)
			.reduceGroup(new RichGroupReduceFunction<Tuple2<String, Double>, Tuple2<String, Double>>() {
			public void reduce(Iterable<Tuple2<String, Double>> values, Collector<Tuple2<String, Double>> out) {}
		}).name("reducer")
		.output(new DiscardingOutputFormat<Tuple2<String, Double>>()).name("sink");
		
		Plan p = env.createProgramPlan();
		OptimizedPlan op = compileNoStats(p);
		
		OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
		
		// get the original nodes
		SourcePlanNode sourceNode = resolver.getNode("source");
		SingleInputPlanNode reduceNode = resolver.getNode("reducer");
		SinkPlanNode sinkNode = resolver.getNode("sink");
		
		// check wiring
		assertEquals(sourceNode, reduceNode.getInput().getSource());
		assertEquals(reduceNode, sinkNode.getInput().getSource());
		
		// check that both reduce and combiner have the same strategy
		assertEquals(DriverStrategy.SORTED_GROUP_REDUCE, reduceNode.getDriverStrategy());
		
		// check the keys
		assertEquals(new FieldList(1), reduceNode.getKeys(0));
		assertEquals(new FieldList(1), reduceNode.getInput().getLocalStrategyKeys());
		
		// check parallelism
		assertEquals(6, sourceNode.getParallelism());
		assertEquals(8, reduceNode.getParallelism());
		assertEquals(8, sinkNode.getParallelism());
	}
	catch (Exception e) {
		System.err.println(e.getMessage());
		e.printStackTrace();
		fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
	}
}

开发者ID:axbaretto，项目名称:flink，代码行数:49，代码来源:GroupReduceCompilationTest.java

示例6: testGroupedReduceWithSelectorFunctionKeyNoncombinable

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
@Test
public void testGroupedReduceWithSelectorFunctionKeyNoncombinable() {
	try {
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(8);
		
		DataSet<Tuple2<String, Double>> data = env.readCsvFile("file:///will/never/be/read").types(String.class, Double.class)
			.name("source").setParallelism(6);
		
		data
			.groupBy(new KeySelector<Tuple2<String,Double>, String>() { 
				public String getKey(Tuple2<String, Double> value) { return value.f0; }
			})
			.reduceGroup(new RichGroupReduceFunction<Tuple2<String, Double>, Tuple2<String, Double>>() {
			public void reduce(Iterable<Tuple2<String, Double>> values, Collector<Tuple2<String, Double>> out) {}
		}).name("reducer")
		.output(new DiscardingOutputFormat<Tuple2<String, Double>>()).name("sink");
		
		Plan p = env.createProgramPlan();
		OptimizedPlan op = compileNoStats(p);
		
		OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
		
		// get the original nodes
		SourcePlanNode sourceNode = resolver.getNode("source");
		SingleInputPlanNode reduceNode = resolver.getNode("reducer");
		SinkPlanNode sinkNode = resolver.getNode("sink");
		
		// get the key extractors and projectors
		SingleInputPlanNode keyExtractor = (SingleInputPlanNode) reduceNode.getInput().getSource();
		SingleInputPlanNode keyProjector = (SingleInputPlanNode) sinkNode.getInput().getSource();
		
		// check wiring
		assertEquals(sourceNode, keyExtractor.getInput().getSource());
		assertEquals(keyProjector, sinkNode.getInput().getSource());
		
		// check that both reduce and combiner have the same strategy
		assertEquals(DriverStrategy.SORTED_GROUP_REDUCE, reduceNode.getDriverStrategy());
		
		// check the keys
		assertEquals(new FieldList(0), reduceNode.getKeys(0));
		assertEquals(new FieldList(0), reduceNode.getInput().getLocalStrategyKeys());
		
		// check parallelism
		assertEquals(6, sourceNode.getParallelism());
		assertEquals(6, keyExtractor.getParallelism());
		
		assertEquals(8, reduceNode.getParallelism());
		assertEquals(8, keyProjector.getParallelism());
		assertEquals(8, sinkNode.getParallelism());
	}
	catch (Exception e) {
		System.err.println(e.getMessage());
		e.printStackTrace();
		fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
	}
}

开发者ID:axbaretto，项目名称:flink，代码行数:58，代码来源:GroupReduceCompilationTest.java

示例7: testAllGroupReduceNoCombiner

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
@Test
public void testAllGroupReduceNoCombiner() {
	try {
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		env.setDegreeOfParallelism(8);
		
		DataSet<Double> data = env.fromElements(0.2, 0.3, 0.4, 0.5).name("source");
		
		data.reduceGroup(new RichGroupReduceFunction<Double, Double>() {
			public void reduce(Iterable<Double> values, Collector<Double> out) {}
		}).name("reducer")
		.print().name("sink");
		
		Plan p = env.createProgramPlan();
		OptimizedPlan op = compileNoStats(p);
		
		OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
		
		
		// the all-reduce has no combiner, when the DOP of the input is one
		
		SourcePlanNode sourceNode = resolver.getNode("source");
		SingleInputPlanNode reduceNode = resolver.getNode("reducer");
		SinkPlanNode sinkNode = resolver.getNode("sink");
		
		// check wiring
		assertEquals(sourceNode, reduceNode.getInput().getSource());
		assertEquals(reduceNode, sinkNode.getInput().getSource());
		
		// check that reduce has the right strategy
		assertEquals(DriverStrategy.ALL_GROUP_REDUCE, reduceNode.getDriverStrategy());
		
		// check DOP
		assertEquals(1, sourceNode.getDegreeOfParallelism());
		assertEquals(1, reduceNode.getDegreeOfParallelism());
		assertEquals(1, sinkNode.getDegreeOfParallelism());
	}
	catch (Exception e) {
		System.err.println(e.getMessage());
		e.printStackTrace();
		fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
	}
}

开发者ID:citlab，项目名称:vs.msc.ws14，代码行数:44，代码来源:GroupReduceCompilationTest.java

示例8: testAllReduceWithCombiner

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
@Test
public void testAllReduceWithCombiner() {
	try {
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		env.setDegreeOfParallelism(8);
		
		DataSet<Long> data = env.generateSequence(1, 8000000).name("source");
		
		GroupReduceOperator<Long, Long> reduced = data.reduceGroup(new RichGroupReduceFunction<Long, Long>() {
			public void reduce(Iterable<Long> values, Collector<Long> out) {}
		}).name("reducer");
		
		reduced.setCombinable(true);
		reduced.print().name("sink");
		
		Plan p = env.createProgramPlan();
		OptimizedPlan op = compileNoStats(p);
		
		OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
		
		// get the original nodes
		SourcePlanNode sourceNode = resolver.getNode("source");
		SingleInputPlanNode reduceNode = resolver.getNode("reducer");
		SinkPlanNode sinkNode = resolver.getNode("sink");
		
		// get the combiner
		SingleInputPlanNode combineNode = (SingleInputPlanNode) reduceNode.getInput().getSource();
		
		// check wiring
		assertEquals(sourceNode, combineNode.getInput().getSource());
		assertEquals(reduceNode, sinkNode.getInput().getSource());
		
		// check that both reduce and combiner have the same strategy
		assertEquals(DriverStrategy.ALL_GROUP_REDUCE, reduceNode.getDriverStrategy());
		assertEquals(DriverStrategy.ALL_GROUP_COMBINE, combineNode.getDriverStrategy());
		
		// check DOP
		assertEquals(8, sourceNode.getDegreeOfParallelism());
		assertEquals(8, combineNode.getDegreeOfParallelism());
		assertEquals(1, reduceNode.getDegreeOfParallelism());
		assertEquals(1, sinkNode.getDegreeOfParallelism());
	}
	catch (Exception e) {
		System.err.println(e.getMessage());
		e.printStackTrace();
		fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
	}
}

开发者ID:citlab，项目名称:vs.msc.ws14，代码行数:49，代码来源:GroupReduceCompilationTest.java

示例9: testGroupedReduceWithFieldPositionKeyNonCombinable

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
@Test
public void testGroupedReduceWithFieldPositionKeyNonCombinable() {
	try {
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		env.setDegreeOfParallelism(8);
		
		DataSet<Tuple2<String, Double>> data = env.readCsvFile("file:///will/never/be/read").types(String.class, Double.class)
			.name("source").setParallelism(6);
		
		data
			.groupBy(1)
			.reduceGroup(new RichGroupReduceFunction<Tuple2<String, Double>, Tuple2<String, Double>>() {
			public void reduce(Iterable<Tuple2<String, Double>> values, Collector<Tuple2<String, Double>> out) {}
		}).name("reducer")
		.print().name("sink");
		
		Plan p = env.createProgramPlan();
		OptimizedPlan op = compileNoStats(p);
		
		OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
		
		// get the original nodes
		SourcePlanNode sourceNode = resolver.getNode("source");
		SingleInputPlanNode reduceNode = resolver.getNode("reducer");
		SinkPlanNode sinkNode = resolver.getNode("sink");
		
		// check wiring
		assertEquals(sourceNode, reduceNode.getInput().getSource());
		assertEquals(reduceNode, sinkNode.getInput().getSource());
		
		// check that both reduce and combiner have the same strategy
		assertEquals(DriverStrategy.SORTED_GROUP_REDUCE, reduceNode.getDriverStrategy());
		
		// check the keys
		assertEquals(new FieldList(1), reduceNode.getKeys(0));
		assertEquals(new FieldList(1), reduceNode.getInput().getLocalStrategyKeys());
		
		// check DOP
		assertEquals(6, sourceNode.getDegreeOfParallelism());
		assertEquals(8, reduceNode.getDegreeOfParallelism());
		assertEquals(8, sinkNode.getDegreeOfParallelism());
	}
	catch (Exception e) {
		System.err.println(e.getMessage());
		e.printStackTrace();
		fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
	}
}

开发者ID:citlab，项目名称:vs.msc.ws14，代码行数:49，代码来源:GroupReduceCompilationTest.java

示例10: testGroupedReduceWithFieldPositionKeyCombinable

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
@Test
public void testGroupedReduceWithFieldPositionKeyCombinable() {
	try {
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		env.setDegreeOfParallelism(8);
		
		DataSet<Tuple2<String, Double>> data = env.readCsvFile("file:///will/never/be/read").types(String.class, Double.class)
			.name("source").setParallelism(6);
		
		GroupReduceOperator<Tuple2<String, Double>, Tuple2<String, Double>> reduced = data
				.groupBy(1)
				.reduceGroup(new RichGroupReduceFunction<Tuple2<String, Double>, Tuple2<String, Double>>() {
			public void reduce(Iterable<Tuple2<String, Double>> values, Collector<Tuple2<String, Double>> out) {}
		}).name("reducer");
		
		reduced.setCombinable(true);
		reduced.print().name("sink");
		
		Plan p = env.createProgramPlan();
		OptimizedPlan op = compileNoStats(p);
		
		OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
		
		// get the original nodes
		SourcePlanNode sourceNode = resolver.getNode("source");
		SingleInputPlanNode reduceNode = resolver.getNode("reducer");
		SinkPlanNode sinkNode = resolver.getNode("sink");
		
		// get the combiner
		SingleInputPlanNode combineNode = (SingleInputPlanNode) reduceNode.getInput().getSource();
		
		// check wiring
		assertEquals(sourceNode, combineNode.getInput().getSource());
		assertEquals(reduceNode, sinkNode.getInput().getSource());
		
		// check that both reduce and combiner have the same strategy
		assertEquals(DriverStrategy.SORTED_GROUP_REDUCE, reduceNode.getDriverStrategy());
		assertEquals(DriverStrategy.SORTED_GROUP_COMBINE, combineNode.getDriverStrategy());
		
		// check the keys
		assertEquals(new FieldList(1), reduceNode.getKeys(0));
		assertEquals(new FieldList(1), combineNode.getKeys(0));
		assertEquals(new FieldList(1), combineNode.getKeys(1));
		assertEquals(new FieldList(1), reduceNode.getInput().getLocalStrategyKeys());
		
		// check DOP
		assertEquals(6, sourceNode.getDegreeOfParallelism());
		assertEquals(6, combineNode.getDegreeOfParallelism());
		assertEquals(8, reduceNode.getDegreeOfParallelism());
		assertEquals(8, sinkNode.getDegreeOfParallelism());
	}
	catch (Exception e) {
		System.err.println(e.getMessage());
		e.printStackTrace();
		fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
	}
}

开发者ID:citlab，项目名称:vs.msc.ws14，代码行数:58，代码来源:GroupReduceCompilationTest.java

示例11: testGroupedReduceWithSelectorFunctionKeyNoncombinable

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
@Test
public void testGroupedReduceWithSelectorFunctionKeyNoncombinable() {
	try {
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		env.setDegreeOfParallelism(8);
		
		DataSet<Tuple2<String, Double>> data = env.readCsvFile("file:///will/never/be/read").types(String.class, Double.class)
			.name("source").setParallelism(6);
		
		data
			.groupBy(new KeySelector<Tuple2<String,Double>, String>() { 
				public String getKey(Tuple2<String, Double> value) { return value.f0; }
			})
			.reduceGroup(new RichGroupReduceFunction<Tuple2<String, Double>, Tuple2<String, Double>>() {
			public void reduce(Iterable<Tuple2<String, Double>> values, Collector<Tuple2<String, Double>> out) {}
		}).name("reducer")
		.print().name("sink");
		
		Plan p = env.createProgramPlan();
		OptimizedPlan op = compileNoStats(p);
		
		OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
		
		// get the original nodes
		SourcePlanNode sourceNode = resolver.getNode("source");
		SingleInputPlanNode reduceNode = resolver.getNode("reducer");
		SinkPlanNode sinkNode = resolver.getNode("sink");
		
		// get the key extractors and projectors
		SingleInputPlanNode keyExtractor = (SingleInputPlanNode) reduceNode.getInput().getSource();
		SingleInputPlanNode keyProjector = (SingleInputPlanNode) sinkNode.getInput().getSource();
		
		// check wiring
		assertEquals(sourceNode, keyExtractor.getInput().getSource());
		assertEquals(keyProjector, sinkNode.getInput().getSource());
		
		// check that both reduce and combiner have the same strategy
		assertEquals(DriverStrategy.SORTED_GROUP_REDUCE, reduceNode.getDriverStrategy());
		
		// check the keys
		assertEquals(new FieldList(0), reduceNode.getKeys(0));
		assertEquals(new FieldList(0), reduceNode.getInput().getLocalStrategyKeys());
		
		// check DOP
		assertEquals(6, sourceNode.getDegreeOfParallelism());
		assertEquals(6, keyExtractor.getDegreeOfParallelism());
		
		assertEquals(8, reduceNode.getDegreeOfParallelism());
		assertEquals(8, keyProjector.getDegreeOfParallelism());
		assertEquals(8, sinkNode.getDegreeOfParallelism());
	}
	catch (Exception e) {
		System.err.println(e.getMessage());
		e.printStackTrace();
		fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
	}
}

开发者ID:citlab，项目名称:vs.msc.ws14，代码行数:58，代码来源:GroupReduceCompilationTest.java

示例12: testGroupedReduceWithSelectorFunctionKeyCombinable

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
@Test
public void testGroupedReduceWithSelectorFunctionKeyCombinable() {
	try {
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		env.setDegreeOfParallelism(8);
		
		DataSet<Tuple2<String, Double>> data = env.readCsvFile("file:///will/never/be/read").types(String.class, Double.class)
			.name("source").setParallelism(6);
		
		GroupReduceOperator<Tuple2<String, Double>, Tuple2<String, Double>> reduced = data
			.groupBy(new KeySelector<Tuple2<String,Double>, String>() { 
				public String getKey(Tuple2<String, Double> value) { return value.f0; }
			})
			.reduceGroup(new RichGroupReduceFunction<Tuple2<String, Double>, Tuple2<String, Double>>() {
			public void reduce(Iterable<Tuple2<String, Double>> values, Collector<Tuple2<String, Double>> out) {}
		}).name("reducer");
		
		reduced.setCombinable(true);
		reduced.print().name("sink");
		
		Plan p = env.createProgramPlan();
		OptimizedPlan op = compileNoStats(p);
		
		OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(op);
		
		// get the original nodes
		SourcePlanNode sourceNode = resolver.getNode("source");
		SingleInputPlanNode reduceNode = resolver.getNode("reducer");
		SinkPlanNode sinkNode = resolver.getNode("sink");
		
		// get the combiner
		SingleInputPlanNode combineNode = (SingleInputPlanNode) reduceNode.getInput().getSource();
		
		// get the key extractors and projectors
		SingleInputPlanNode keyExtractor = (SingleInputPlanNode) combineNode.getInput().getSource();
		SingleInputPlanNode keyProjector = (SingleInputPlanNode) sinkNode.getInput().getSource();
		
		// check wiring
		assertEquals(sourceNode, keyExtractor.getInput().getSource());
		assertEquals(keyProjector, sinkNode.getInput().getSource());
		
		// check that both reduce and combiner have the same strategy
		assertEquals(DriverStrategy.SORTED_GROUP_REDUCE, reduceNode.getDriverStrategy());
		assertEquals(DriverStrategy.SORTED_GROUP_COMBINE, combineNode.getDriverStrategy());
		
		// check the keys
		assertEquals(new FieldList(0), reduceNode.getKeys(0));
		assertEquals(new FieldList(0), combineNode.getKeys(0));
		assertEquals(new FieldList(0), combineNode.getKeys(1));
		assertEquals(new FieldList(0), reduceNode.getInput().getLocalStrategyKeys());
		
		// check DOP
		assertEquals(6, sourceNode.getDegreeOfParallelism());
		assertEquals(6, keyExtractor.getDegreeOfParallelism());
		assertEquals(6, combineNode.getDegreeOfParallelism());
		
		assertEquals(8, reduceNode.getDegreeOfParallelism());
		assertEquals(8, keyProjector.getDegreeOfParallelism());
		assertEquals(8, sinkNode.getDegreeOfParallelism());
	}
	catch (Exception e) {
		System.err.println(e.getMessage());
		e.printStackTrace();
		fail(e.getClass().getSimpleName() + " in test: " + e.getMessage());
	}
}

开发者ID:citlab，项目名称:vs.msc.ws14，代码行数:67，代码来源:GroupReduceCompilationTest.java

示例13: checkCombinability

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
private void checkCombinability() {
	if (function instanceof FlatCombineFunction &&
			function.getClass().getAnnotation(RichGroupReduceFunction.Combinable.class) != null) {
		this.combinable = true;
	}
}

开发者ID:citlab，项目名称:vs.msc.ws14，代码行数:7，代码来源:GroupReduceOperator.java

示例14: translateSelectorFunctionDistinct

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
private static <IN, OUT, K> PlanUnwrappingReduceGroupOperator<IN, OUT, K> translateSelectorFunctionDistinct(
		Keys.SelectorFunctionKeys<IN, ?> rawKeys, RichGroupReduceFunction<IN, OUT> function,
		TypeInformation<IN> inputType, TypeInformation<OUT> outputType, String name, Operator<IN> input)
{
	@SuppressWarnings("unchecked")
	final Keys.SelectorFunctionKeys<IN, K> keys = (Keys.SelectorFunctionKeys<IN, K>) rawKeys;
	
	TypeInformation<Tuple2<K, IN>> typeInfoWithKey = new TupleTypeInfo<Tuple2<K, IN>>(keys.getKeyType(), inputType);
	
	KeyExtractingMapper<IN, K> extractor = new KeyExtractingMapper<IN, K>(keys.getKeyExtractor());


	PlanUnwrappingReduceGroupOperator<IN, OUT, K> reducer =
			new PlanUnwrappingReduceGroupOperator<IN, OUT, K>(function, keys, name, outputType, typeInfoWithKey, true);
	
	MapOperatorBase<IN, Tuple2<K, IN>, MapFunction<IN, Tuple2<K, IN>>> mapper = new MapOperatorBase<IN, Tuple2<K, IN>, MapFunction<IN, Tuple2<K, IN>>>(extractor, new UnaryOperatorInformation<IN, Tuple2<K, IN>>(inputType, typeInfoWithKey), "Key Extractor");

	reducer.setInput(mapper);
	mapper.setInput(input);
	
	// set the mapper's parallelism to the input parallelism to make sure it is chained
	mapper.setDegreeOfParallelism(input.getDegreeOfParallelism());
	
	return reducer;
}

开发者ID:citlab，项目名称:vs.msc.ws14，代码行数:26，代码来源:DistinctOperator.java

示例15: TupleUnwrappingFlatCombinableGroupReducer

import org.apache.flink.api.common.functions.RichGroupReduceFunction; //导入依赖的package包/类
private TupleUnwrappingFlatCombinableGroupReducer(RichGroupReduceFunction<IN, OUT> wrapped) {
	super(wrapped);
	this.iter = new TupleUnwrappingIterator<IN, K>();
	this.coll = new TupleWrappingCollector<IN, K>(this.iter);
}

开发者ID:citlab，项目名称:vs.msc.ws14，代码行数:6，代码来源:PlanUnwrappingReduceGroupOperator.java

注：本文中的org.apache.flink.api.common.functions.RichGroupReduceFunction类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Java ICommonsMap类代码示例发布时间：2022-05-22

Java PropertyPath类代码示例发布时间：2022-05-22

剪的笔顺,诠释剪的笔画,认识剪的部首

1 六六分期app的软件客服如何联系？(六六分期

六六分期app的软件客服如何联系？不知道吗？加qq群【895510560】即可！标题：六六分期

阅读：19284|2023-10-27

2 可心卡盟:win10系统火狐flash插件崩溃怎么

今天小编告诉大家如何处理win10系统火狐flash插件总是崩溃的问题，可能很多用户都不知

阅读：10021|2022-11-06

3 亲亲特价:怎么删除回收站图标

今天小编告诉大家如何对win10系统删除桌面回收站图标进行设置，可能很多用户都不知道

阅读：8344|2022-11-06

4 济南大学虚拟社区:鲁大师节能降温的具体办

今天小编告诉大家如何对win10系统电脑设置节能降温的设置方法，想必大家都遇到过需要

阅读：8713|2022-11-06

5 xlueops.exe:无线网络安装向导

我们在使用xp系统的过程中,经常需要对xp系统无线网络安装向导设置进行设置，可能很多

阅读：8658|2022-11-06

6 女斗合众国:win7系统cf与主机连接不稳定怎

今天小编告诉大家如何处理win7系统玩cf老是与主机连接不稳定的问题，可能很多用户都不

阅读：9688|2022-11-06

7 0xc000022-[cf烟雾头]cf怎么调烟雾头

电脑对日常生活的重要性小编就不多说了，可是一旦碰到win7系统设置cf烟雾头的问题，很

阅读：8647|2022-11-06

8 qizideyouhuo:应用程序无法正常启动0xc0000

我们在日常使用电脑的时候，有的小伙伴们可能在打开应用的时候会遇见提示应用程序无法

阅读：8014|2022-11-06

9 ipz-185:win7系统vcf文件怎么打开

今天小编告诉大家如何对win7系统打开vcf文件进行设置，可能很多用户都不知道怎么对win

阅读：8683|2022-11-06

10 傻哥蹦迪:win10系统s4怎么打开usb调试

今天小编告诉大家如何对win10系统s4开启USB调试模式进行设置，可能很多用户都不知道怎

阅读：7550|2022-11-06

客服电话

电子邮件

Java RichGroupReduceFunction类代码示例

示例1: testBasicType

示例2: testBasicType

示例3: PlanUnwrappingReduceGroupOperator

示例4: testAllGroupReduceNoCombiner

示例5: testGroupedReduceWithFieldPositionKeyNonCombinable

示例6: testGroupedReduceWithSelectorFunctionKeyNoncombinable

示例7: testAllGroupReduceNoCombiner

示例8: testAllReduceWithCombiner

示例9: testGroupedReduceWithFieldPositionKeyNonCombinable

示例10: testGroupedReduceWithFieldPositionKeyCombinable

示例11: testGroupedReduceWithSelectorFunctionKeyNoncombinable

示例12: testGroupedReduceWithSelectorFunctionKeyCombinable

示例13: checkCombinability

示例14: translateSelectorFunctionDistinct

示例15: TupleUnwrappingFlatCombinableGroupReducer

请发表评论

全部评论

上一篇：

下一篇：

微信小程序websocket+node.js(ws)

librespeed/speedtest: Self-hosted Speedt

avehtari/BDA_m_demos: Bayesian Data Anal

四维彩超怎么看性别？四维看男孩女孩诀窍

medfreeman/markdown-it-toc-and-anchor: m

剪的笔顺,诠释剪的笔画,认识剪的部首

六六分期app的软件客服如何联系？(六六分期

florent37/ViewAnimator: A fluent Android

florent37/Shrine-MaterialDesign2: implem

CVE-2020-36276

SimpleSoftwareIO/simple-sms: Send and re

关于我们

产品与服务

解决方案

139-2527-9053