...
The figure below summarizes the iteration paradigm described above. The streams in red color indicates inputs provided by the caller to the iteration body, as well as outputs emitted by the iteration body to the caller.
Public Interfaces
We propose to make the following API changes to support the iteration paradigm described above.
...
This interface abstracts the core computation logic that is expected to be iteratively computed. Any algorithm, which wants to apply iterative computation with feedback streams, should implement this interface.
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
package org.apache.flink.ml.iteration; @PublicEvolving public interface IterationBody { /** * This method creates the graph for the iteration body. * * See Utils::iterate, Utils::iterateBoundedStreams and Utils::iterateAndReplayBoundedStreams for how the iteration * body can be executed and when execution of the corresponding graph should terminate. * * Required: the number of feedback variable streams returned by this method must equal the number of variable * streams given to this method. * * @param variableStreams the variable streams. * @param dataStreams the data streams. * @return a IterationBodyResult. */ IterationBodyResult process(DataStreamList variableStreams, DataStreamList dataStreams); } |
...
This interface allows users to register callbacks that can be used to a) run an algorithm in sync mode; and b) emit final output after the iteration terminates.
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
org.apache.flink.ml.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 increments. The initial epoch watermark is -1. * * 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 IterationUtils 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 by 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 IterationUtils 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); } } |
...
The common and the primary functionality of APIs in this class is this: Union the feedback variables streams (returned by the iteration body) with the initial variable streams (provided by the user) and use the merged streams as inputs to the iteration body.
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
package org.apache.flink.ml.iteration; /** * A helper class to apply {@link IterationBody} to data streams. */ @PublicEvolving public class IterationUtils { /** * This method can use an iteration body to process records in unbounded data streams. * * This method invokes the iteration body with the following parameters: * 1) The 1st 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). * 2) The 2nd parameter is the data streams given to this method. * * The epoch values are determined as described below. See IterationListener for how the epoch values are used. * 1) All records in the initial variable streams has epoch=0. * 2) All records in the data streams has epoch=MAX_LONG. In this case, records in the data stream won't affect * any operator's low watermark. * 3) 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 onEpochLWIncremented(), then * the epoch of this record = incrementedEpochLW - 1. * 4) For any record emitted by this operator into a feedback variable stream, the epoch of the emitted record = * min(the epoch of the input record that triggers this emission, MAX_LONG - 1) + 1. If this record is emitted by * onEpochLWIncremented(), then the epoch of this record = incrementedEpochLW. * * The execution of the graph created by the iteration body will not terminate by itself. This is because at least * one of its data streams is unbounded. * * Required: * 1) All the init variable streams must be bounded. * 2) There is at least one unbounded stream in the data streams list. * 3) The parallelism of any stream in the initial variable streams must equal the parallelism of the stream at the * same index of the feedback variable streams returned by the IterationBody. * * @param initVariableStreams The initial variable streams. These streams will be merged with the feedback variable * streams before being used as the 1st parameter to invoke the iteration body. * @param dataStreams The data streams. These streams will be used as the 2nd parameter to invoke the iteration * body. * @param body The computation logic which takes variable/data streams and returns variable/output streams. * @return The list of output streams returned by the iteration boy. */ static DataStreamList iterateUnboundedStreams(DataStreamList initVariableStreams, DataStreamList dataStreams, IterationBody body) {...} /** * This method can use 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. * * This method invokes the iteration body with the following parameters: * 1) The 1st 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). * 2) The 2nd parameter is the data streams given to this method. * * The epoch values are determined as described below. See IterationListener for how the epoch values are used. * 1) All records in the initial variable streams has epoch=0. * 2) All records in the data streams has epoch=0. * 3) 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 onEpochLWIncremented(), then * the epoch of this record = incrementedEpochLW - 1. * 4) 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. If this record is emitted by onEpochLWIncremented(), * then the epoch of this record = incrementedEpochLW. * * Suppose there is a coordinator operator which takes all feedback variable streams (emitted by the iteration body) * and the termination criteria stream (if not null) as inputs. The execution of the graph created by the * iteration body will terminate when all input streams have been fully AND any of the following conditions is met: * consumed: * 1) The termination criteria stream is not null. And the coordinator operator has not observed any new value from * the termination criteria stream between two consecutive onEpochLWIncremented invocations. * 2) The coordinator operator has not observed any new value from any feedback variable stream between two * consecutive onEpochLWIncremented invocations. * * Required: * 1) All the init variable streams and the data streams must be bounded. * 2) The parallelism of any stream in the initial variable streams must equal the parallelism of the stream at the * same index of the feedback variable streams returned by the IterationBody. * * @param initVariableStreams The initial variable streams. These streams will be merged with the feedback variable * streams before being used as the 1st parameter to invoke the iteration body. * @param dataStreams The data streams. These streams will be used as the 2nd parameter to invoke the iteration * body. * @param body The computation logic which takes variable/data streams and returns variable/output streams. * @return The list of output streams returned by the iteration boy. */ static DataStreamList iterateBoundedStreamsUntilTermination(DataStreamList initVariableStreams, DataStreamList dataStreams, IterationBody body) {...} /** * This method can use 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 replays records in the data streams, the iteration body does not need to cache those * records to visit those records repeatedly. * * This method invokes the iteration body with the following parameters: * 1) The 1st 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). * 2) The 2nd parameter is a list of replayed data streams, which are created by replaying the initial data streams * round by round until the iteration terminates. The records in the Nth round will be emitted into the iteration * body only if the low watermark of the first operator in the iteration body >= N - 1. * * The epoch values are determined as described below. See IterationListener for how the epoch values are used. * 1) All records in the initial variable streams has epoch=0. * 2) The records from the initial data streams will be replayed round by round into the iteration body. The records * in the first round have epoch=0. And records in the Nth round have epoch = N - 1. * 3) 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 onEpochLWIncremented(), then * the epoch of this record = incrementedEpochLW - 1. * 4) For any record emitted by this operator into a feedback stream, the epoch of the emitted record = the epoch * of the input record that triggers this emission + 1. If this record is emitted by onEpochLWIncremented(), then * the epoch of this record = incrementedEpochLW. * * Suppose there is a coordinator operator which takes all feedback variable streams (emitted by the iteration body) * and the termination criteria stream (if not null) as inputs. The execution of the graph created by the * iteration body will terminate when all input streams have been fully consumed AND any of the following conditions * is met: * 1) The termination criteria stream is not null. And the coordinator operator has not observed any new value from * the termination criteria stream between two consecutive onEpochLWIncremented invocations. * 2) The coordinator operator has not observed any new value from any feedback variable stream between two * consecutive onEpochLWIncremented invocations. * * Required: * 1) All the init variable streams and the data streams must be bounded. * 2) The parallelism of any stream in the initial variable streams must equal the parallelism of the stream at the * same index of the feedback variable streams returned by the IterationBody. * * @param initVariableStreams The initial variable streams. These streams will be merged with the feedback variable * streams before being used as the 1st parameter to invoke the iteration body. * @param initDataStreams The initial data streams. Records from these streams will be repeatedly replayed and used * as the 2nd parameter to invoke the iteration body. * @param body The computation logic which takes variable/data streams and returns variable/output streams. * @return The list of output streams returned by the iteration boy. */ static DataStreamList iterateAndReplayBoundedStreamsUntilTermination(DataStreamList initVariableStreams, DataStreamList initDataStreams, IterationBody body) {...} } |
...
IterationBodyResult is a helper class which contains the streams returned by the IterationBody::process(...)
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
package org.apache.flink.ml.iteration; /** * A helper class that contains the streams returned by the iteration body. */ class IterationBodyResult { /** * A list of feedback variable streams. These streams will only be used during the iteration execution and will * not be returned to the caller of the iteration body. It is assumed that the method which executes the * iteration body will feed the records of the feedback variable streams back to the corresponding input variable * streams. */ DataStreamList feedbackVariableStreams; /** * A list of output streams. These streams will be returned to the caller of the methods that execute the * iteration body. */ DataStreamList outputStreams; /** * An optional termination criteria stream. If this stream is not null, it will be used together with the * feedback variable streams to determine when the iteration should terminate. */ Optional<DataStream<?>> terminationCriteria; } |
...
DataStreamList is a helper class that serves as a container for a list of data streams with possibly different elements types.
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
package org.apache.flink.ml.iteration; public class DataStreamList { // 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) {...} } |
...