flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Lydia Ickler <ickle...@googlemail.com>
Subject BulkIteration and BroadcastVariables
Date Wed, 30 Mar 2016 15:33:58 GMT
Hi all,
I have a question regarding the BulkIteration and BroadcastVariables:
The BulkIteration by default has one input variable and sends one variable into the next iteration,
right?
What if I need to collect some intermediate results in each iteration? How would I do that?

For example in my code below I would like to store all newEigenValue. Unfortunately I didn’t
find a way to do so.
Is it possible to set/change BroadcastVariables? Or is it only possible to „get“ them?

Thanks in advance!
Lydia


//read input file
DataSet<Tuple3<Integer, Integer, Double>> matrixA = readMatrix(env, input);


//initial:
//Approximate EigenVector by PowerIteration
DataSet<Tuple3<Integer, Integer, Double>> eigenVector = PowerIteration_getEigenVector2(matrixA);
//Approximate EigenValue by PowerIteration
DataSet<Tuple3<Integer, Integer, Double>> oldEigenValue = PowerIteration_getEigenValue(matrixA,eigenVector);
//Deflate original matrix
matrixA = PowerIteration_getNextMatrix(matrixA,eigenVector,oldEigenValue);

DataSet<Tuple3<Integer, Integer, Double>> newEigenVector = null;
DataSet<Tuple3<Integer, Integer, Double>> newEigenValue = null;
DataSet<Tuple3<Integer, Integer, Double>> newMatrixA = null;


//BulkIteration to find k dominant eigenvalues
IterativeDataSet<Tuple3<Integer, Integer, Double>> iteration = matrixA.iterate(outer_iterations);

newEigenVector = PowerIteration_getEigenVector2(iteration);
newEigenValue = PowerIteration_getEigenValue(iteration,newEigenVector);
newMatrixA = PowerIteration_getNextMatrix(iteration,newEigenVector,newEigenValue);

//get gap
DataSet<Tuple3<Integer, Integer, Double>> gap = newEigenValue.map(new getGap()).withBroadcastSet(oldEigenValue,
"oldEigenValue");
DataSet<Tuple3<Integer, Integer, Double>> filtered = gap.filter(new gapFilter());
oldEigenValue = newEigenValue;

DataSet<Tuple3<Integer, Integer, Double>> neue  = iteration.closeWith(newMatrixA,filtered);


Mime
View raw message