我用flink创建了一个java程序，可以将2个矩阵相乘。我正在使用批处理（dataset）环境来处理它，并且我想显示这个的执行计划。当我做一些流（datastream）的例子时，我刚刚调用了 StreamExecutionEnvironment.getExecutionEnvironment().getExecutionPlan() . flink有同样的方法可用于批处理，但当我调用它时，我得到错误： java.lang.RuntimeException: No new data sinks have been defined since the last execution. The last execution refers to the latest call to 'execute()', 'count()', 'collect()', or 'print()'. . 我想我在做这里描述的事情：https://ci.apache.org/projects/flink/flink-docs-stable/dev/execution_plans.html 但由于某种原因，我得到了例外。
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.MapPartitionFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.util.Collector;
public class MatrixMultiplication {
private static final String MATRIX_A = "A";
private static final String MATRIX_B = "B";
public MatrixMultiplication() throws Exception {
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSet<Tuple4<String, Integer, Integer, Integer>> matrixA = env.readCsvFile("resources/matrixA.csv")
.fieldDelimiter(",").types(Integer.class, Integer.class, Integer.class)
.map(t -> new Tuple4<String, Integer, Integer, Integer>("A", t.f0, t.f1, t.f2))
.returns(Types.TUPLE(Types.STRING, Types.INT, Types.INT, Types.INT));
System.out.println("Matrix A");
matrixA.print();
DataSet<Tuple4<String, Integer, Integer, Integer>> matrixB = env.readCsvFile("resources/matrixB.csv")
.fieldDelimiter(",").types(Integer.class, Integer.class, Integer.class)
.map(t -> new Tuple4<String, Integer, Integer, Integer>("B", t.f0, t.f1, t.f2))
.returns(Types.TUPLE(Types.STRING, Types.INT, Types.INT, Types.INT));
System.out.println("Matrix B");
matrixB.print();
int columnsMatrixB = 2;
int linesMatrixA = 2;
DataSet<Tuple2<Tuple3<Integer, Integer, Integer>, Integer>> keyValueMatrixA = matrixA
.mapPartition(new MapMatrixToKeysAndValues(columnsMatrixB));
DataSet<Tuple2<Tuple3<Integer, Integer, Integer>, Integer>> keyValueMatrixB = matrixB
.mapPartition(new MapMatrixToKeysAndValues(linesMatrixA));
DataSet<Tuple2<Tuple3<Integer, Integer, Integer>, Integer>> matrixAB = keyValueMatrixA.union(keyValueMatrixB);
DataSet<Tuple2<Tuple3<Integer, Integer, Integer>, Integer>> matrixAB_01 = matrixAB.groupBy(0)
.reduce(new ProductReducer());
DataSet<Tuple2<Tuple2<Integer, Integer>, Integer>> matrixAB_02 = matrixAB_01.map(new SumMapper());
DataSet<Tuple2<Tuple2<Integer, Integer>, Integer>> productMatrixAB = matrixAB_02.groupBy(0)
.reduce(new SumReducer());
System.out.println("Matrix AB");
productMatrixAB.print();
// String executionPlan = env.getExecutionPlan();
// System.out.println("ExecutionPlan ........................ ");
System.out.println(productMatrixAB.getExecutionEnvironment().getExecutionPlan());
// System.out.println("........................ ");
}
public static class MapMatrixToKeysAndValues implements
MapPartitionFunction<Tuple4<String, Integer, Integer, Integer>, Tuple2<Tuple3<Integer, Integer, Integer>, Integer>> {
private static final long serialVersionUID = 6992353073599144457L;
private int count;
public MapMatrixToKeysAndValues(int count) {
this.count = count;
}
@Override
public void mapPartition(Iterable<Tuple4<String, Integer, Integer, Integer>> values,
Collector<Tuple2<Tuple3<Integer, Integer, Integer>, Integer>> out) throws Exception {
for (Tuple4<String, Integer, Integer, Integer> tuple : values) {
for (int c = 1; c <= count; c++) {
Tuple3<Integer, Integer, Integer> key = null;
Integer value = null;
if (MATRIX_A.equals(tuple.f0)) {
// key(i,k,i+j) for k=1...N
Integer i = tuple.f1;
Integer j = tuple.f2;
Integer k = c;
key = new Tuple3<Integer, Integer, Integer>(i, k, i + j);
// value matrix[i,j]
value = tuple.f3;
} else if (MATRIX_B.equals(tuple.f0)) {
// key(i,k,i+j) for i=1...L
Integer i = c;
Integer j = tuple.f1;
Integer k = tuple.f2;
key = new Tuple3<Integer, Integer, Integer>(i, k, i + j);
// value matrix[j,k]
value = tuple.f3;
}
out.collect(new Tuple2<Tuple3<Integer, Integer, Integer>, Integer>(key, value));
}
}
}
}
public static class ProductReducer implements ReduceFunction<Tuple2<Tuple3<Integer, Integer, Integer>, Integer>> {
private static final long serialVersionUID = 6166767956669902083L;
@Override
public Tuple2<Tuple3<Integer, Integer, Integer>, Integer> reduce(
Tuple2<Tuple3<Integer, Integer, Integer>, Integer> value1,
Tuple2<Tuple3<Integer, Integer, Integer>, Integer> value2) throws Exception {
Integer product = null;
product = value1.f1 * value2.f1;
return new Tuple2<Tuple3<Integer, Integer, Integer>, Integer>(value1.f0, product);
}
}
public static class SumMapper implements
MapFunction<Tuple2<Tuple3<Integer, Integer, Integer>, Integer>, Tuple2<Tuple2<Integer, Integer>, Integer>> {
private static final long serialVersionUID = -1437482917757334157L;
@Override
public Tuple2<Tuple2<Integer, Integer>, Integer> map(Tuple2<Tuple3<Integer, Integer, Integer>, Integer> value)
throws Exception {
Tuple2<Integer, Integer> key = new Tuple2<Integer, Integer>(value.f0.f0, value.f0.f1);
return new Tuple2<Tuple2<Integer, Integer>, Integer>(key, value.f1);
}
}
public static class SumReducer implements ReduceFunction<Tuple2<Tuple2<Integer, Integer>, Integer>> {
private static final long serialVersionUID = 7849401047616065465L;
@Override
public Tuple2<Tuple2<Integer, Integer>, Integer> reduce(Tuple2<Tuple2<Integer, Integer>, Integer> value1,
Tuple2<Tuple2<Integer, Integer>, Integer> value2) throws Exception {
Tuple2<Integer, Integer> key = new Tuple2<Integer, Integer>(value1.f0.f0, value1.f0.f1);
Integer value = value1.f1 + value2.f1;
return new Tuple2<Tuple2<Integer, Integer>, Integer>(key, value);
}
}
}$cat资源/matrixa.csv
1,1,1
1,2,3
1,3,4
1,4,-2
2,1,6
2,2,2
2,3,-3
2,4,1$cat资源/matrixb.csv
1,1,1
1,2,-2
2,1,4
2,2,3
3,1,-3
3,2,-2
4,1,0
4,2,4
1条答案
按热度按时间ct3nt3jp1#
如果您删除了
print()电话。print()也会触发执行。