fsk119 commented on code in PR #26694: URL: https://github.com/apache/flink/pull/26694#discussion_r2157917013
########## docs/content/docs/dev/table/sql/queries/model-inference.md: ########## @@ -0,0 +1,138 @@ +--- +title: "Model Inference" +weight: 7 +type: docs +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Model Inference + +{{< label Streaming >}} + +Flink SQL provides the `ML_PREDICT` table-valued function (TVF) to perform model inference in SQL queries. This function allows you to apply machine learning models to your data streams directly in SQL. + +## ML_PREDICT Function + +The `ML_PREDICT` function takes a table input, applies a model to it, and returns a new table with the model's predictions. The function supports both synchronous and asynchronous inference modes. + +### Syntax + +```sql +ML_PREDICT( Review Comment: How about using this case? Users may think ml_predict is a scalar result if we don't give the whole example. SELECT* FROM ML_PREDICT(...) ########## docs/content/docs/dev/table/sql/create.md: ########## @@ -890,3 +891,46 @@ Language tag to instruct Flink runtime how to execute the function. Currently on Specifies the list of jar resources that contain the implementation of the function along with its dependencies. The jar should be located in a local or remote [file system]({{< ref "docs/deployment/filesystems/overview" >}}) such as hdfs/s3/oss which Flink current supports. <span class="label label-danger">Attention</span> Currently only JAVA, SCALA language support USING clause. + +{{< top >}} + +## CREATE MODEL +```sql +CREATE [TEMPORARY] MODEL [IF NOT EXISTS] [catalog_name.][db_name.]model_name + [( + { <input_column_definition> }[ , ...n] + { <output_column_definition> }[ , ...n] + )] + [COMMENT model_comment] + WITH (key1=val1, key2=val2, ...) + +<input_column_definition>: + column_name column_type [COMMENT column_comment] + +<output_column_definition>: + column_name column_type [COMMENT column_comment] +``` + +Create a model with optional input and output column definitions. If a model with the same name already exists in the catalog, an exception is thrown. + +**TEMPORARY** + +Create a temporary model that has catalog and database namespaces and overrides models. + +**IF NOT EXISTS** + +If the model already exists, nothing happens. + +**Input/Output Columns** + +The input columns define the features that will be used for model inference. The output columns define the predictions that the model will produce. Each column must have a name and data type. The input and output columns Review Comment: It's better we can remove the description about the model tranning because we still doesn't have any plan or FLIP yet. I am not sure whether we will do this in the near future. ########## docs/content/docs/dev/table/sql/create.md: ########## @@ -890,3 +891,46 @@ Language tag to instruct Flink runtime how to execute the function. Currently on Specifies the list of jar resources that contain the implementation of the function along with its dependencies. The jar should be located in a local or remote [file system]({{< ref "docs/deployment/filesystems/overview" >}}) such as hdfs/s3/oss which Flink current supports. <span class="label label-danger">Attention</span> Currently only JAVA, SCALA language support USING clause. + +{{< top >}} + +## CREATE MODEL +```sql +CREATE [TEMPORARY] MODEL [IF NOT EXISTS] [catalog_name.][db_name.]model_name + [( + { <input_column_definition> }[ , ...n] + { <output_column_definition> }[ , ...n] + )] + [COMMENT model_comment] + WITH (key1=val1, key2=val2, ...) + +<input_column_definition>: + column_name column_type [COMMENT column_comment] + +<output_column_definition>: + column_name column_type [COMMENT column_comment] +``` + +Create a model with optional input and output column definitions. If a model with the same name already exists in the catalog, an exception is thrown. + +**TEMPORARY** + +Create a temporary model that has catalog and database namespaces and overrides models. + +**IF NOT EXISTS** + +If the model already exists, nothing happens. + +**Input/Output Columns** + +The input columns define the features that will be used for model inference. The output columns define the predictions that the model will produce. Each column must have a name and data type. The input and output columns +are optional because it can be inferred from training data later if Flink supports model training. For existing models, they are required to be defined. + +**WITH OPTIONS** + +Model properties used to store extra information related to this model. The properties are usually used to find and create the underlying model provider. +The key and value of expression `key1=val1` should both be string literal. + +**Note:** The model properties and supported model types may vary depending on the underlying model provider. + Review Comment: It's better we can an example to illustrate how to create a model with sql. ########## docs/content/docs/dev/table/sql/queries/model-inference.md: ########## @@ -0,0 +1,138 @@ +--- +title: "Model Inference" +weight: 7 +type: docs +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Model Inference + +{{< label Streaming >}} + +Flink SQL provides the `ML_PREDICT` table-valued function (TVF) to perform model inference in SQL queries. This function allows you to apply machine learning models to your data streams directly in SQL. + +## ML_PREDICT Function + +The `ML_PREDICT` function takes a table input, applies a model to it, and returns a new table with the model's predictions. The function supports both synchronous and asynchronous inference modes. Review Comment: he function offers support for synchronous/asynchronous inference modes when the underlying model permits both. ########## docs/content/docs/dev/table/sql/queries/model-inference.md: ########## @@ -0,0 +1,138 @@ +--- +title: "Model Inference" +weight: 7 +type: docs +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Model Inference + +{{< label Streaming >}} + +Flink SQL provides the `ML_PREDICT` table-valued function (TVF) to perform model inference in SQL queries. This function allows you to apply machine learning models to your data streams directly in SQL. + +## ML_PREDICT Function + +The `ML_PREDICT` function takes a table input, applies a model to it, and returns a new table with the model's predictions. The function supports both synchronous and asynchronous inference modes. + +### Syntax + +```sql +ML_PREDICT( + TABLE input_table, + MODEL model_name, + DESCRIPTOR(feature_columns), + [CONFIG => MAP['key', 'value']] +) +``` + +### Parameters + +- `input_table`: The input table containing the data to be processed +- `model_name`: The name of the model to use for inference +- `feature_columns`: A descriptor specifying which columns from the input table should be used as features for the model +- `config`: (Optional) A map of configuration options for the model inference + +### Configuration Options + +The following configuration options can be specified in the config map: + +- `async`: (Optional) Boolean value indicating whether to use asynchronous inference. +- `timeout`: (Optional) Timeout duration for asynchronous inference (e.g., '100s') +- `capacity`: (Optional) Buffer capacity for asynchronous inference + +Default configuration values: +- [table.exec.async-ml-predict.buffer-capacity]({{< ref "docs/dev/table/config/#table-exec-async-ml-predict-buffer-capacity" >}}) +- [table.exec.async-ml-predict.output-mode]({{< ref "docs/dev/table/config/#table-exec-async-ml-predict-output-mode" >}}) +- [table.exec.async-ml-predict.timeout]({{< ref "docs/dev/table/config/#table-exec-async-ml-predict-timeout" >}}) + +### Example + +```sql +-- Basic usage +SELECT * FROM ML_PREDICT( + TABLE input_table, + MODEL my_model, + DESCRIPTOR(feature1, feature2) +); + +-- With configuration options +SELECT * FROM ML_PREDICT( + TABLE input_table, + MODEL my_model, + DESCRIPTOR(feature1, feature2), + CONFIG => MAP['async', 'true', 'timeout', '100s'] +); + +-- Using named parameters +SELECT * FROM ML_PREDICT( + INPUT => TABLE input_table, + MODEL => MODEL my_model, + ARGS => DESCRIPTOR(feature1, feature2), + CONFIG => MAP['async', 'true'] +); +``` + +### Output + +The output table contains all columns from the input table plus the model's prediction columns. The prediction columns are added based on the model's output schema. + +### Notes + +1. The model must be registered in the catalog before it can be used with `ML_PREDICT`. +2. The number of feature columns specified in the descriptor must match the model's input schema. +3. If column names in the output conflict with existing column names in the input table, an index will be added to the output column names to avoid conflicts. For example, if the output column is named `prediction`, it will be renamed to `prediction0` if a column with that name already exists in the input table. +4. For asynchronous inference, the model provider must support the `AsyncPredictRuntimeProvider` interface. + +### Model Provider + +The `ML_PREDICT` function uses a `ModelProvider` to perform the actual model inference. The provider is looked up based on the provider identifier specified when registering the model. There are two types of model providers: + +1. `PredictRuntimeProvider`: For synchronous model inference + - Implements the `createPredictFunction` method to create a synchronous prediction function + - Used when `async` is set to `false` in the config + +2. `AsyncPredictRuntimeProvider`: For asynchronous model inference + - Implements the `createAsyncPredictFunction` method to create an asynchronous prediction function + - Used when `async` is set to `true` in the config + - Requires additional configuration for timeout and buffer capacity + +If `async` is not set in the config, the system will pick either sync or async model provider and prefer async model provider if both exist. + +### Error Handling + +The function will throw an exception in the following cases: +- The model does not exist in the catalog +- The number of feature columns does not match the model's input schema +- The model parameter is missing +- Too few or too many arguments are provided + +### Performance Considerations + +1. For high-throughput scenarios, consider using asynchronous inference mode. +2. Configure appropriate timeout and buffer capacity values for asynchronous inference. +3. The function's performance depends on the underlying model provider implementation. + +### Related Functions Review Comment: Related Statements -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
