Status
Current state: "Under Discussion"
Discussion thread: To be added
...
In order to address the issues described above and support iteration on both bounded and unbounded data streams, this FLIP proposes to add a couple APIs in the flink-ml repository. Note that we have chosen to put the iteration API (and its implementation) in the flink-ml repository instead of the DataStream class in the Flink core repository, because we believe it is important to keep the Flink core runtime as simple and maintainable as possible.
...
In the following, we describe the example use-cases and the exact semantics that we aim to support via the proposed APIs.
...
Terminology
We expect to support online training and offline training use-cases. These use-cases have the following properties:1) In online trainingexplain a few terminologies in the following to facilitate the reading of this doc.
1) Iteration body
An iteration body is a subgraph of a Flink job graph that implements the iteration logic of e.g. a machine learning algorithm. In particular, the iteration body will create feedback variable streams and emit values into those streams. The feedback variable streams will be joined with the user-provided initial variable streams and the resulting streams will be ingested into the iteration body.
Target use-cases
Different algorithms might have different requirements for the input datasets (bounded or unbounded), synchronization between parallel subtasks (sync or async), amount of data processed for every variable update (a batch/subset or the entire dataset). We describe each of these requirements below.
1) Algorithms have different needs for whether the input data streams should be bounded or unbounded. We classify those algorithms into online algorithm and offline algorithms as below.
- For online training algorithms, the training samples will be unbounded streams of data. The corresponding iteration body should ingest these unbounded streams of data, read each value in each stream once, and update machine learning model repeatedly in near real-time. The iteration will never terminate in this case. The
...
- algorithm should be
...
- executed as a streaming job.
...
- For offline training algorithms, the training samples will be bounded streams of data. The corresponding iteration body should read these bounded data streams for arbitrary number of rounds and update machine learning model repeatedly until a termination criteria is met (e.g. a given number of rounds is reached or the model has converged). The
...
- algorithm should be
...
- executed as a batch job.
For both online training and offline training scenarios, there exists use-cases to run the iteration body in either sync mode or async mode:
...
2) Algorithms (either online or offline) have different needs of how their parallel subtasks.
- In the sync mode, parallel subtasks, which execute the iteration body, update the model variables in a coordinated manner. There exists global epoch epochs, such that all subtasks
...
- read the shared model variables at the beginning of
...
- an epoch, calculate variable updates based on the fetched variable values, and
...
- write updates of the variable values at the end of this epoch
...
- .
...
...
- In the async mode, each parallel subtask, which execute the iteration body, could read/update the shared model variables without waiting for variable updates from other subtask. For example, a subtask could have updated model variables 10 times when another subtask has updated model variables only 3 times.
The sync mode is useful when an algorithm should be executed in a deterministic way to achieve best possible accuracy, and the straggler issue (i.e. there is subtask which is considerably slower than others) does not cause slow down the algorithm execution too much. In comparison, the async mode is useful for users who to execute an algorithm algorithms which want to be parallelized and executed across many subtasks as fast as possible, in the presence of without worrying about performance issue caused by stragglers, at the possible cost of reduced accuracy.
3) An algorithm may have additional requirements in how much data should be consumed each time before a subtask can update variables. There are two categories of choices here:
- The algorithm wants to update variables every time an arbitrary subset of the user-provided data streams (either bounded or unbounded) is processed.
- The algorithm wants to update variables every time all data of the user-provided bounded data streams is processed.
In the machine learning domain, some algorithms allow users to configure a batch size and the model will be updated every time each subtask processes a batch of data. Those algorithms fits into the first category. And such an algorithm can be either online or offline.
Other algorithms only update variables every time the entire data is consumed for one round. Those algorithms fit into the second category. And such an algorithm must be offline because, by this definition, the user-provided dataset must be bounded.
Public Interfaces
We propose to add the following interfaces and utility classes in the flink-ml repository.
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
public class DataStreamList { public DataStreamList(DataStream<?>... dataStreams) {...} // 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) {...} } @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); } /** * 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; } /** * 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); } } /** * 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 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 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 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) {...} } |