Status
Current state: "Under Discussion"
...
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Table of Contents |
---|
Motivation and Use-cases
The existing Flink ML library allows users to compose an Estimator/Transformer from a pipeline (i.e. linear sequence) of Estimator/Transformer, and each Estimator/Transformer has one input and one output.
...
7) There is no API provided by the Estimator/Transformer interface to validate the schema consistency of a Pipeline. Users would have to instantiate Tables (with I/O logics) and run fit/transform to know whether the stages in the Pipeline are compatible with each other.
Public Interfaces
This FLIP proposes quite a few changes and addition to the existing Flink ML APIs. We first describe the proposed API additions and changes, followed by the API code of interfaces and classes after making the proposed changes.
API additions and changes
Here we list the additions and the changes to the Flink ML API.
...
This change addresses the use-cases described in the motivation section, where we need to compose an Estimator from a DAG of Estimator/Transformer.
Interfaces and classes after the proposed API changes
The following code block shows the interface of Stage, Transformer, Estimator, Pipeline and PipelineModel after the proposed changes.
...
Code Block | ||
---|---|---|
| ||
/** * A Graph acts as an Estimator. It consists of a DAG of stages, each of which is either an * Estimator or Transformer. */ @PublicEvolving public final class Graph implements Estimator<Graph, GraphModel> { public Graph(...) {...} @Override public GraphModel fit(Table... inputs) {...} @Override public TableSchema[] transformSchemas(TableSchema... schemas) { return schemas; } /** Skipped a few methods, including the implementations of some Estimator APIs. */ } /** A GraphModel acts as a Transformer. It consists of a DAG of Transformers. */ @PublicEvolving public final class GraphModel implements Transformer<GraphModel> { /** Skipped a few methods, including the implementations of the Transformer APIs. */ } /** A GraphBuilder helps connect Stage instances into a Graph or GraphModel. */ @PublicEvolving public final class GraphBuilder { /** * Specifies the upper bound (could be loose) of the number of output tables that can be * returned by the Transformer::getStateStreams and Transformer::transform methods, for any * stage involved in this Graph. * * <p>The default upper bound is 20. */ public GraphBuilder setMaxOutputLength(int maxOutputLength) {...} /** * Creates a TableId associated with this GraphBuilder. It can be used to specify the passing of * tables between stages, as well as the input/output tables of the Graph/GraphModel generated * by this builder. */ public TableId createTableId() {...} /** * The Graph::fit and GraphModel::transform should invoke the fit/transform of the corresponding * stage with the corresponding inputs. * * <p>Returns a list of TableIds, which represents outputs of the Transformer::transform * invocation. */ public TableId[] getOutputs(Stage<?> stage, TableId... inputs) {...} /** * The GraphModel::setStateStreams should invoke the setStateStreams of the corresponding stage * with the corresponding inputs. */ void setStateStreams(Stage<?> stage, TableId... inputs) {...} /** * The GraphModel::getStateStreams should invoke the getStateStreams of the corresponding stage. * * <p>Returns a list of TableIds, which represents outputs of the getStateStreams invocation. */ TableId[] getStateStreams(Stage<?> stage) {...} /** * Returns a Graph instance which the following API specification: - Graph::fit should take * inputs and returns a GraphModel with the following specification. - GraphModel::transform * should take inputs and returns outputs. - GraphModel::setStateStreams should take * inputStates. - GraphModel::getStateStreams should return outputStates. * * <p>The fit/transform/setStateStreams/getStateStreams should invoke the APIs of the internal * stages in the order specified by the DAG of stages. */ Graph build(TableId[] inputs, TableId[] outputs, TableId[] inputStates, TableId[] outputStates) {...} /** * Returns a GraphModel instance which the following API specification: - GraphModel::transform * should take inputs and returns outputs. - GraphModel::setStateStreams should take * inputStates. - GraphModel::getStateStreams should return outputStates. * * <p>The transform/setStateStreams/getStateStreams should invoke the APIs of the internal * stages in the order specified by the DAG of stages. * * <p>This method throws exception if any stage of this graph is an Estimator. */ GraphModel buildModel(TableId[] inputs, TableId[] outputs, TableId[] inputStates, TableId[] outputStates) {...} // The TableId is necessary to pass the inputs/outputs of various API calls across the // Graph/GraphModel stags. static class TableId {} } |
Example Usage
In this section we provide examples code snippets to demonstrate how we can use the APIs proposed in this FLIP to address the use-cases in the motivation section.
Composing an Estimator from a DAG of Estimator/Transformer
Suppose we have the following Transformer and Estimator classes:
...
Code Block | ||
---|---|---|
| ||
GraphBuilder builder = new GraphBuilder(); // Creates nodes Stage<?> stage1 = new TransformerA(); Stage<?> stage2 = new TransformerA(); Stage<?> stage3 = new EstimatorB(); // Creates inputs and inputStates TableId input1 = builder.createTableId(); TableId input2 = builder.createTableId(); // Feeds inputs to nodes and gets outputs. TableId output1 = builder.getOutputs(stage1, input1)[0]; TableId output2 = builder.getOutputs(stage2, input2)[0]; TableId output3 = builder.getOutputs(stage3, output1, output2)[0]; // Specifies the ordered lists of inputs, outputs, input states and output states that will // be used as the inputs/outputs of the corresponding Graph and GraphModel APIs. TableId[] inputs = new TableId[] {input1, input2}; TableId[] outputs = new TableId[] {output3}; // Generates the Graph instance. Graph graph = builder.build(inputs, outputs, new TableId[]{}, new TableId[]{}); // Use the Graph instance as an Estimator. GraphModel model = graph.fit(...); Table[] results = model.transform(...); |
Online learning by running Transformer and Estimator concurrently on different machines
Here is an online learning scenario:
...
Code Block | ||
---|---|---|
| ||
void runInferenceOnWebServer(...) { // Reads the transformer_json from the same remote file written by the above code snippet. String transformer_json = ...; // Creates the state stream from Kafka topicA which is written by the above code snippet. Table state_stream = ...; // Creates the input stream that needs inference. Table input_stream = ...; Transformer transformer = new Transformer(...); transformer.loadJson(transformer_json); transformer.setStateStreams(new Table[]{state_stream}); Table output_stream = transformer.transform(input_stream); // Do something with the output_stream. // Executes the operators generated by the Transformer::transform(...), which reads from state_stream to update its parameters. It also does inference on input_stream and produces results to the output_stream. env.execute() } |
Compatibility, Deprecation, and Migration Plan
The changes proposed in this FLIP is backward incompatible with the existing APIs. We propose to change the APIs directly without deprecation period. And we will manually migrate the existing open source projects which use the existing Flink ML API to use the proposed APIs.
...
To our best knowledge, the only open source project that uses the Flink ML API is https://github.com/alibaba/Alink. We will work together with Alink developers to migrate the existing code to use the proposed API. Furthermore, we will migrate Alink's Estimator/Transformer implementation to the Flink ML library codebase as much as possible.
Test Plan
We will provide unit tests to validate the proposed changes.
Rejected Alternatives
There is no rejected alternatives to be listed here yet.
...