...
...
...
...
...
JIRA: <TODO>
This method is easier to use, but it also limit some possible optimizations
...
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
-
[This FLIP proposal is a joint work between Yun Gao, Dong Lin and Zhipeng Zhang]
...
It is important to note that the users typically should not invoke the IterationBody::process directly because the model-variables expected by the iteration body is not the same as the initial-model-variables provided by the user. Instead, model-variables are computed as the union of the feedback-model-variables (emitted by the iteration body) and the initial-model-variables (provided by the caller of the iteration body). To relieve user from creating this union operator, we have added utility class (see IterationUtilsIterations) to run an iteration-body with the user-provided inputs.
...
Code Block |
---|
language | java |
---|
linenumbers | true |
---|
|
package org.apache.flink.iteration
/**
* The callbacks defined below will be invoked only if the operator instance which implements this interface is used
* within an iteration body.
*/
@PublicEvolving
public interface IterationListener<T> {
/**
* This callback is invoked every time the epoch watermark of this operator increments. The initial epoch watermark
* of an operator is 0.
*
* The epochWatermark is the maximum integer that meets this requirement: every record that arrives at the operator
* going forward should have an epoch larger than the epochWatermark. See Java docs in IterationUtilsIterations for how epoch
* is determined for records ingested into the iteration body and for records emitted by operators within the
* iteration body.
*
* If all inputs are bounded, the maximum epoch of all records ingested into this operator is used as the
* epochWatermark parameter for the last invocation of this callback.
*
* @param epochWatermark The incremented epoch watermark.
* @param context A context that allows emitting side output. The context is only valid during the invocation of
* this method.
* @param collector The collector for returning result values.
*/
void onEpochWatermarkIncremented(int epochWatermark, Context context, Collector<T> collector);
/**
* This callback is invoked after the execution of the iteration body has terminated.
*
* See Java doc of methods in IterationUtilsIterations for the termination conditions.
*
* @param context A context that allows emitting side output. The context is only valid during the invocation of
* this method.
* @param collector The collector for returning result values.
*/
void onIterationTermination(Context context, Collector<T> collector);
/**
* Information available in an invocation of the callbacks defined in the IterationProgressListener.
*/
interface Context {
/**
* Emits a record to the side output identified by the {@link OutputTag}.
*
* @param outputTag the {@code OutputTag} that identifies the side output to emit to.
* @param value The record to emit.
*/
<X> void output(OutputTag<X> outputTag, X value);
}
} |
...
Code Block |
---|
language | java |
---|
linenumbers | true |
---|
|
package org.apache.flink.iteration;
/**
* A helper class to create iterations. To construct an iteration, Users are required to provide
*
* <ul>
* <li>initVariableStreams: the initial values of the variable data streams which would be updated
* in each round.
* <li>dataStreams: the other data streams used inside the iteration, but would not be updated.
* <li>iterationBody: specifies the subgraph to update the variable streams and the outputs.
* </ul>
*
* <p>The iteration body will be invoked with two parameters: The first parameter is a list of input
* variable streams, which are created as the union of the initial variable streams and the
* corresponding feedback variable streams (returned by the iteration body); The second parameter is
* the data streams given to this method.
*
* <p>During the execution of iteration body, each of the records involved in the iteration has an
* epoch attached, which is mark the progress of the iteration. The epoch is computed as:
*
* <ul>
* <li>All records in the initial variable streams and initial data streams has epoch = 0.
* <li>For any record emitted by this operator into a non-feedback stream, the epoch of this
* emitted record = the epoch of the input record that triggers this emission. If this record
* is emitted by onEpochWatermarkIncremented(), then the epoch of this record =
* epochWatermark.
* <li>For any record emitted by this operator into a feedback variable stream, the epoch of the
* emitted record = the epoch of the input record that triggers this emission + 1.
* </ul>
*
* <p>The framework would given the notification at the end of each epoch for operators and UDFs
* that implements {@link IterationListener}.
*
* <p>The limitation of constructing the subgraph inside the iteration body could be refer in {@link
* IterationBody}.
*
* <p>An example of the iteration is like:
*
* <pre>{@code
* DataStreamList result = Iterations.iterateUnboundedStreams(
* DataStreamList.of(first, second),
* DataStreamList.of(third),
* (variableStreams, dataStreams) -> {
* ...
* return new IterationBodyResult(
* DataStreamList.of(firstFeedback, secondFeedback),
* DataStreamList.of(output));
* }
* result.<Integer>get(0).addSink(...);
* }</pre>
*/
public class Iterations {
/**
* This method uses an iteration body to process records in possibly unbounded data streams. The
* iteration would not terminate if at least one of its inputs is unbounded. Otherwise it will
* terminated after all the inputs are terminated and no more records are iterating.
*
* @param initVariableStreams The initial variable streams, which is merged with the feedback
* variable streams before being used as the 1st parameter to invoke the iteration body.
* @param dataStreams The non-variable streams also refered in the {@code body}.
* @param body The computation logic which takes variable/data streams and returns
* feedback/output streams.
* @return The list of output streams returned by the iteration boy.
*/
public static DataStreamList iterateUnboundedStreams(
DataStreamList initVariableStreams, DataStreamList dataStreams, IterationBody body) {
return null;
}
/**
* This method uses an iteration body to process records in some bounded data streams
* iteratively until a termination criteria is reached (e.g. the given number of rounds is
* completed or no further variable update is needed). Because this method does not replay
* records in the data streams, the iteration body needs to cache those records in order to
* visit those records repeatedly.
*
* @param initVariableStreams The initial variable streams, which is merged with the feedback
* variable streams before being used as the 1st parameter to invoke the iteration body.
* @param dataStreams The non-variable streams also refered in the {@code body}.
* @param config The config for the iteration, like whether to re-create the operator on each
* round.
* @param body The computation logic which takes variable/data streams and returns
* feedback/output streams.
* @return The list of output streams returned by the iteration boy.
*/
public static DataStreamList iterateBoundedStreamsUntilTermination(
DataStreamList initVariableStreams,
ReplayableDataStreamList dataStreams,
IterationConfig config,
IterationBody body) {
return null;
}
} |
5) Introduce the PerRoundSubGraphBuilder forEachRound utility method.
PerRoundSubgraphBuilder forEachRound allows the users to specify a sub-graph that executes as per-round mode, namely all the operators would be re-created for each round.
Code Block |
---|
|
package org.apache.flink.iteration;
import org.apache.flink.annotation.Experimental;
/** Constructs a subgraph inside the iteration body to executepublic interface IterationBody {
....
/**
* @param inputs The inputs of the subgraph.
* @param perRoundSubBody The computational logic that want to be executed as per-round.
*/
@Experimental
public class PerRoundSubGraphBuilder {
@return The output of the subgraph.
*/
static DataStreamList forEachRound(
DataStreamList inputs, PerRoundSubBody perRoundSubBody) {
return null;
}
/** The sub-graph inside the iteration body that should be executed as per-round. */
public interface PerRoundSubGraphPerRoundSubBody {
DataStreamList process(DataStreamList input);
}
public static DataStreamList forEachRound(DataStreamList inputs, PerRoundSubGraph subGraph) {
return null;
}
}} |
6) Add the DataStreamList and ReplayableDataStreamList class.
...
Code Block |
---|
language | java |
---|
linenumbers | true |
---|
|
package org.apache.flink.iteration;
public class DataStreamList {
public static DataStreamList of(DataStream<?>... streams);
// Returns the number of data streams in this list.
public int size() {...}
// Returns the data stream at the given index in this list.
public <T> DataStream<T> get(int index) {...}
}
public class ReplayableDataStreamList {
public static ReplayableDataStreamList ofprivate final List<DataStream<?>> replayedDataStreams;
private final List<DataStream<?>> nonReplayedStreams;
private ReplayableDataStreamList(
Tuple2<DataStream<List<DataStream<?>>> replayedDataStreams, Boolean>... dataStreamAndIfNeedReplays);
List<DataStream<?>> nonReplayedStreams) {
public static Tuple2<DataStream<?>, Boolean> replay(DataStream<?> dataStream);
this.replayedDataStreams = replayedDataStreams;
public static Tuple2<DataStream<?>, Boolean> noReplay(DataStream<?> dataStream);
/** Returns the number of data streams in this list. */
public int size();
/** Returns the data stream at the given index in this list. */
@SuppressWarnings("unchecked")
public <T> DataStream<T> get(int index);
public boolean shouldReplay(int index); this.nonReplayedStreams = nonReplayedStreams;
}
public static ReplayedDataStreamList replay(DataStream<?>... dataStreams) {
return new ReplayedDataStreamList(Arrays.asList(dataStreams));
}
public static NonReplayedDataStreamList notReplay(DataStream<?>... dataStreams) {
return new NonReplayedDataStreamList(Arrays.asList(dataStreams));
}
List<DataStream<?>> getReplayedDataStreams() {
return Collections.unmodifiableList(replayedDataStreams);
}
List<DataStream<?>> getNonReplayedStreams() {
return Collections.unmodifiableList(nonReplayedStreams);
}
private static class ReplayedDataStreamList extends ReplayableDataStreamList {
public ReplayedDataStreamList(List<DataStream<?>> replayedDataStreams) {
super(replayedDataStreams, Collections.emptyList());
}
public ReplayableDataStreamList andNotReplay(DataStream<?>... nonReplayedStreams) {
return new ReplayableDataStreamList(
getReplayedDataStreams(), Arrays.asList(nonReplayedStreams));
}
}
private static class NonReplayedDataStreamList extends ReplayableDataStreamList {
public NonReplayedDataStreamList(List<DataStream<?>> nonReplayedDataStreams) {
super(Collections.emptyList(), nonReplayedDataStreams);
}
public ReplayableDataStreamList andReplay(DataStream<?>... replayedStreams) {
return new ReplayableDataStreamList(
Arrays.asList(replayedStreams), getNonReplayedStreams());
}
}
} |
7) Introduce the IterationConfig
...
Code Block |
---|
|
package org.apache.flink.iteration;
/** The config for an iteration. */
public class IterationConfig {
private final OperatorLifeCycle operatorRoundMode;
publicprivate IterationConfig(OperatorLifeCycle operatorRoundMode) {
this.operatorRoundMode = operatorRoundMode;
}
public static IterationConfigBuilder newBuilder() {
return new IterationConfigBuilder();
}
public static class IterationConfigBuilder {
private OperatorLifeCycle operatorRoundMode = OperatorLifeCycle.ALL_ROUND;
private IterationConfigBuilder() {}
public IterationConfigBuilder setOperatorRoundMode(OperatorLifeCycle operatorRoundMode) {
this.operatorRoundMode = operatorRoundMode;
return this;
}
public IterationConfig build() {
return new IterationConfig(operatorRoundMode);
}
}
public enum OperatorLifeCycle {
ALL_ROUND,
PER_ROUND
}
} |
...
Code Block |
---|
language | java |
---|
linenumbers | true |
---|
|
public class SynchronousBoundedLinearRegression {
private static final int N_DIM = 50;
private static final int N_EPOCH = 5;
private static final int N_BATCH_PER_EPOCH = 10;
private static final OutputTag<double[]> FINAL_MODEL_OUTPUT_TAG = new OutputTag<double[]>{};
public static void main(String[] args) {
DataStream<double[]> initParameters = loadParameters().setParallelism(1);
DataStream<Tuple2<double[], Double>> dataset = loadDataSet().setParallelism(1);
DataStreamList resultStreams =
IterationUtilsIterations.iterateBoundedStreamsUntilTermination(
DataStreamList.of(initParameters),
ReplayableDataStreamList.ofnotReplay(noReplay(dataset)),
IterationConfig.newBuilder().setOperatorRoundMode(ALL_ROUND).build();
(variableStreams, dataStreams) -> {
DataStream<double[]> parameterUpdates = variableStreams.get(0);
DataStream<Tuple2<double[], Double>> dataset = dataStreams.get(0);
SingleOutputStreamOperator<double[]> parameters = parameterUpdates.process(new ParametersCacheFunction());
DataStream<double[]> modelUpdate = parameters.setParallelism(1)
.broadcast()
.connect(dataset)
.coProcess(new TrainFunction())
.setParallelism(10)
.process(new ReduceFunction())
.setParallelism(1)
return new IterationBodyResult(DataStreamList.of(modelUpdate), DataStreamList.of(parameters.getSideOut(FINAL_MODEL_OUTPUT_TAG)));
});
DataStream<double[]> finalModel = resultStreams.get("final_model");
finalModel.print();
}
public static class ParametersCacheFunction extends ProcessFunction<double[], double[]>
implements IterationListener<double[]> {
private final double[] parameters = new double[N_DIM];
public void processElement(double[] update, Context ctx, Collector<O> output) {
// Suppose we have a util to add the second array to the first.
ArrayUtils.addWith(parameters, update);
}
void onEpochWatermarkIncremented(int epochWatermark, Context context, Collector<T> collector) {
if (epochWatermark < N_EPOCH * N_BATCH_PER_EPOCH) {
collector.collect(parameters);
}
}
public void onIterationEnd(int[] round, Context context) {
context.output(FINAL_MODEL_OUTPUT_TAG, parameters);
}
}
public static class TrainFunction extends CoProcessFunction<double[], Tuple2<double[], Double>, double[]> implements IterationListener<double[]> {
private final List<Tuple2<double[], Double>> dataset = new ArrayList<>();
private double[] firstRoundCachedParameter;
private Supplier<int[]> recordRoundQuerier;
public void setCurrentRecordRoundsQuerier(Supplier<int[]> querier) {
this.recordRoundQuerier = querier;
}
public void processElement1(double[] parameter, Context context, Collector<O> output) {
int[] round = recordRoundQuerier.get();
if (round[0] == 0) {
firstRoundCachedParameter = parameter;
return;
}
calculateModelUpdate(parameter, output);
}
public void processElement2(Tuple2<double[], Double> trainSample, Context context, Collector<O> output) {
dataset.add(trainSample)
}
void onEpochWatermarkIncremented(int epochWatermark, Context context, Collector<T> collector) {
if (epochWatermark == 0) {
calculateModelUpdate(firstRoundCachedParameter, output);
firstRoundCachedParameter = null;
}
}
private void calculateModelUpdate(double[] parameters, Collector<O> output) {
List<Tuple2<double[], Double>> samples = sample(dataset);
double[] modelUpdate = new double[N_DIM];
for (Tuple2<double[], Double> record : samples) {
double diff = (ArrayUtils.muladd(record.f0, parameters) - record.f1);
ArrayUtils.addWith(modelUpdate, ArrayUtils.multiply(record.f0, diff));
}
output.collect(modelUpdate);
}
}
public static class ReduceFunction {
private double[] mergedValue = ArrayUtils.newArray(N_DIM);
public void processElement(double[] parameter, Context context, Collector<O> output) {
mergedValue = ArrayUtils.add(mergedValue, parameter);
}
void onEpochWatermarkIncremented(int epochWatermark, Context context, Collector<T> collector) {
collector.collect(mergedValue);
mergedValue = ArrayUtils.newArray(N_DIM);
}
}
} |
...
Code Block |
---|
|
DataStreamList resultStreams =
IterationUtilsIterations.iterateBoundedStreamsUntilTermination(
DataStreamList.of(initParameters),
ReplayableDataStreamList.of(noReplaynotReplay(dataset)),
IterationConfig.newBuilder().setOperatorRoundMode(ALL_ROUND).build();
(variableStreams, dataStreams) -> {
DataStream<double[]> parameterUpdates = variableStreams.get(0);
DataStream<Tuple2<double[], Double>> dataset = dataStreams.get(0);
SingleOutputStreamOperator<double[]> parameters = parameterUpdates.process(new ParametersCacheFunction());
DataStream<double[]> modelUpdate = parameters.setParallelism(1)
.broadcast()
.connect(dataset)
.coProcess(new TrainFunction())
.setParallelism(10)
DataStream<double[]> reduced = PerRoundGraphBuilder.forEachRound(DataStreamList.of(modelUpdate), streams -> {
return streams.<double[]>get(0).windowAll().reduce((x, y) -> ArrayUtils.add(x, y));
}).<double[]>get(0);
return new IterationBodyResult(DataStreamList.of(modelUpdate), DataStreamList.of(parameters.getSideOut(FINAL_MODEL_OUTPUT_TAG)));
}); |
...
Code Block |
---|
|
public static class ParametersCacheFunction extends ProcessFunction<Tuple2<Integer, double[]>, Tuple2<Integer, double[]>>
implements BoundedIterationProgressListener<double[]> {
private final double[] parameters = new double[N_DIM];
public void processElement(Tuple2<Integer, double[]> update, Context ctx, Collector<Tuple2<Integer, double[]>> output) {
// Suppose we have a util to add the second array to the first.
ArrayUtils.addWith(parameters, update);
output.collect(new Tuple2<>(update.f0, parameters))
}
public void onIterationEnd(int[] round, Context context) {
context.output(FINAL_MODEL_OUTPUT_TAG, parameters);
}
}
public class AsynchronousBoundedLinearRegression {
...
DataStreamList resultStreams =
IterationUtilsIterations.iterateBoundedStreamsUntilTermination(DataStreamList.of(initParameters), DataStreamList.of(dataset), (variableStreams, dataStreams) -> {
DataStream<double> parameterUpdates = variableStreams.get(0);
DataStream<Tuple2<double[], Double>> dataset = dataStreams.get(0);
SingleOutputStreamOperator<double[]> parameters = parameterUpdates.process(new ParametersCacheFunction());
DataStream<double[]> modelUpdate = parameters.setParallelism(1)
.partitionCustom((key, numPartitions) -> key % numPartitions, update -> update.f0)
.connect(dataset)
.coProcess(new TrainFunction())
.setParallelism(10)
return new IterationBodyResult(DataStreamList.of(modelUpdate), DataStreamList.of(parameters.getSideOut(FINAL_MODEL_OUTPUT_TAG)));
});
...
} |
...
Code Block |
---|
|
public class SynchronousUnboundedLinearRegression {
private static final N_DIM = 50;
private static final OutputTag<double[]> MODEL_UPDATE_OUTPUT_TAG = new OutputTag<double[]>{};
public static void main(String[] args) {
DataStream<double[]> initParameters = loadParameters().setParallelism(1);
DataStream<Tuple2<double[], Double>> dataset = loadDataSet().setParallelism(1);
DataStreamList resultStreams = IterationUtilsIterations.iterateUnboundedStreams(DataStreamList.of(initParameters), DataStreamList.of(dataset), (variableStreams, dataStreams) -> {
DataStream<double> parameterUpdates = variableStreams.get(0);
DataStream<Tuple2<double[], Double>> dataset = dataStreams.get(0);
SingleOutputStreamOperator<double[]> parameters = model.process(new ParametersCacheFunction(10));
DataStream<double[]> modelUpdate = parameters.setParallelism(1)
.broadcast()
.connect(dataset)
.transform(
"operator",
BasicTypeInfo.INT_TYPE_INFO,
new TrainOperators(50));
.setParallelism(10);
return new IterationBodyResult(DataStreamList.of(modelUpdate), DataStreamList.of(parameters.getSideOut(FINAL_MODEL_OUTPUT_TAG)));
})
DataStream<double[]> finalModel = resultStreams.get("model_update");
finalModel.addSink(...)
}
public static class ParametersCacheFunction extends ProcessFunction<double[], double[]> {
private final int numOfTrainTasks;
private final int numOfUpdatesReceived = 0;
private final double[] parameters = new double[N_DIM];
public ParametersCacheFunction(int numOfTrainTasks) {
this.numOfTrainTasks = numOfTrainTasks;
}
public void processElement(double[] update, Context ctx, Collector<O> output) {
// Suppose we have a util to add the second array to the first.
ArrayUtils.addWith(parameters, update);
numOfUpdatesReceived++;
if (numOfUpdatesReceived == numOfTrainTasks) {
output.collect(parameters);
numOfUpdatesReceived = 0;
}
}
}
public static class TrainOperators extends AbstractStreamOperator<double[]> implements TwoInputStreamOperator<double[], Tuple2<double[], Double>, double[]>, InputSelectable {
private final int miniBatchSize;
private final List<Tuple2<double[], Double>> miniBatch = new ArrayList<>();
private double[] firstRoundCachedParameter;
public TrainOperators(int miniBatchSize) {
this.miniBatchSize = miniBatchSize;
}
public void processElement1(double[] parameter, Context context, Collector<O> output) {
calculateModelUpdate(parameter, output);
miniBatchSize.clear();
}
public void processElement2(Tuple2<double[], Double> trainSample, Context context, Collector<O> output) {
dataset.add(trainSample);
}
public InputSelection nextSelection() {
if (miniBatch.size() < miniBatchSize) {
return InputSelection.SECOND;
} else {
return InputSelection.FIRST;
}
}
private void calculateModelUpdate(double[] parameters, Collector<O> output) {
double[] modelUpdate = new double[N_DIM];
for (Tuple2<double[], Double> record : miniBatchSize) {
double diff = (ArrayUtils.muladd(record.f0, parameters) - record.f1);
ArrayUtils.addWith(modelUpdate, ArrayUtils.multiply(record.f0, diff));
}
output.collect(modelUpdate);
}
}
} |
...