Call an Agent in Flink SQL (AGENT_RUN)-Realtime Compute for Apache Flink(Flink)-阿里云帮助中心

Realtime Compute for Apache Flink provides the AGENT_RUN table-valued function (TVF) to call Agents built with the Flink Agents framework in Flink SQL jobs. This topic describes the syntax, parameters, examples, and job submission methods of AGENT_RUN.

Prerequisites

An Agent has been built using the Flink Agents framework. For more information, see Developing Flink Agents (Beta).
Realtime Compute engine VVR 11.8 or later is used.

Syntax

SELECT * FROM 
AGENT_RUN(
   TABLE input_table [PARTITION BY key]
  ,AGENT agent_name
 [,LANGUAGE 'java'/'python']
 [,CONFIG => MAP['key', 'value']]
)

Parameters

Parameter	Description	Required
input_table	The input table that contains the data to be processed.	Yes
key	The partition key of the input table. If specified, data with the same key is processed by the same Agent instance.	No
agent_name	The fully qualified class name of the Agent. The Agent must be built using the Flink Agents framework.	Yes
LANGUAGE	The programming language used to develop the Agent. Valid values: `java` and `python`. Default value: `java`.	No
CONFIG	The Flink Agents configuration parameters passed as key-value pairs in MAP format. For more information, see Configuration.	No

Output

The output table contains all columns from the input table, plus the Agent output column. The output column is fixed as output VARIANT.

Examples

Call a Java Agent

CREATE TEMPORARY TABLE reviews
(
    id   INT
    ,msg STRING
)
WITH (
    'connector' = 'datagen'
    ,'number-of-rows' = '3'
    ,'fields.id.kind' = 'sequence'
    ,'fields.id.start' = '1'
    ,'fields.id.end' = '3'
    ,'fields.msg.length' = '4'
);

CREATE TEMPORARY TABLE `test`
(
    id      INT
    ,output VARIANT
)
WITH ('connector' = 'print');

INSERT INTO `test`
SELECT
    id
    ,output
FROM AGENT_RUN(TABLE reviews PARTITION BY id
    ,'org.apache.flink.agents.examples.agents.EchoAgent');

Call a Python Agent

CREATE TEMPORARY TABLE reviews
(
    id   INT
    ,msg STRING
)
WITH (
    'connector' = 'datagen'
    ,'number-of-rows' = '3'
    ,'fields.id.kind' = 'sequence'
    ,'fields.id.start' = '1'
    ,'fields.id.end' = '3'
    ,'fields.msg.length' = '4'
);

CREATE TEMPORARY TABLE `test`
(
    id      INT
    ,output VARIANT
)
WITH ('connector' = 'print');

INSERT INTO `test`
SELECT
    id
    ,output
FROM AGENT_RUN(TABLE reviews PARTITION BY id
    ,'echo_agent.SqlEchoPythonAgent'
    ,'python');

Submit a job

Java Agent job

Upload the JAR file that contains the Agent class as an additional dependency, then submit the SQL job.

Python Agent job

Upload the Python file that contains the Python Agent class as an additional dependency.
Upload the virtual environment (venv) that contains the Flink Agents Python wheel and job-related dependencies. For more information about how to build the venv, see Use a custom Python virtual environment.
Add the following parameters in Runtime Parameters:

Parameter	Description
python.files	The path of the Python file uploaded as an additional dependency. Format: `oss://${bucket}/<filename>.py`. You can find the OSS path on the O&M page.
python.archives	The path of the venv archive that contains Flink Agents and job-related dependencies.
python.client.executable	The Python interpreter path in the uploaded venv, for example, `venv.zip/venv/bin/python`.
python.executable	Fixed value: `python3.10`.
containerized.master.env.FLINK_HOME	Fixed value: `/flink`.
containerized.taskmanager.env.FLINK_HOME	Fixed value: `/flink`.

Example configuration:

    python.files: 'oss://${bucket}/echo_agent.py'
    python.archives: >-
      oss://${bucket}/venv.zip
    python.executable: python3.10
    python.client.executable: venv.zip/venv/bin/python
    containerized.master.env.FLINK_HOME: /flink
    containerized.taskmanager.env.FLINK_HOME: /flink