On this page

Show all

Help Center/ Data Lake Insight/ Flink SQL Syntax Reference/ Flink OpenSource SQL 1.15 Syntax Reference/ Functions/ Parameter Transfer

Parameter Transfer

Updated on 2025-01-10 GMT+08:00

View PDF

Scenario

A UDF can be used in many jobs, and some parameter values vary with jobs. To easily modify the parameter values, you can set pipeline.global-job-parameters in the Runtime Configuration tab on the Flink OpenSource SQL editing page, and then get the parameter values in the UDF code and use the values as you need. You only need to change the parameter values in the runtime configuration tab to pass the new values to the UDF.

Procedure

Use the open(FunctionContext context) method in your UDF to pass parameters through a FunctionContext object. To pass parameters to a job, perform the following steps:

Add pipeline.global-job-parameters to Runtime Configuration on the Flink OpenSource SQL editing page. The format is as follows:

pipeline.global-job-parameters=k1:v1,"k2:v1,v2",k3:"str:ing","k4:str""ing"

This configuration defines a map as shown in Table 1

**Table 1** Examples for pipeline.global-job-parameters
Key	Value
k1	v1
k2	v1,v2
k3	str:ing
k4	str""ing

NOTE:

FunctionContext#getJobParameter obtains only the value of pipeline.global-job-parameters. You need to add all key-value pairs that will be used in the UDF to pipeline.global-job-parameters.
Keys and values are separated by colons (:). All key-values are connected by commas (,).
If the key or value contains commas (,), use double quotation marks (") to enclose key or value, for example, "v1,v2".
If the key or value contains colons (:), use double quotation marks (") to enclose the key or value, for example, "str:ing".
If the key or value contains a double quotation mark("), use another double quotation mark ("") to escape the first one, and use double quotation marks (") to enclose the key or value, for example, "str""ing".

In your UDF code, use FunctionContext#getJobParameter to obtain the key-value pairs you set. The code example is as follows:

context.getJobParameter("url","jdbc:mysql://xx.xx.xx.xx:3306/table");
context.getJobParameter("driver","com.mysql.jdbc.Driver");
context.getJobParameter("user","user");
context.getJobParameter("password","password");

Code Samples

The following sample UDF uses pipeline.global-job-parameters to pass parameters such as url, user, and password required for connecting to the database, obtains the udf_info table data, and combines this data with the stream data into JSON output.

**Table 2** udf_info
key	value
class	class-4

SimpleJsonBuild.java

package udf;

import com.fasterxml.jackson.databind.ObjectMapper;

import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.ScalarFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.HashMap;
import java.util.Map;

public class SimpleJsonBuild extends ScalarFunction {
   private static final Logger LOG = LoggerFactory.getLogger(SimpleJsonBuild.class);
   String remainedKey;
   String remainedValue;

   private Connection initConnection(Map<String, String> userParasMap) {
       String url = userParasMap.get("url");
       String driver = userParasMap.get("driver");
       String user = userParasMap.get("user");
       String password = userParasMap.get("password");
       Connection conn = null;
       try {
           Class.forName(driver);
           conn = DriverManager.getConnection(url, user, password);
           LOG.info("connect successfully");
      } catch (Exception e) {
           LOG.error(String.valueOf(e));
      }
       return conn;
  }

   @Override
   public void open(FunctionContext context) throws Exception {
       Map<String, String> userParasMap = new HashMap<>();
       Connection connection;
       PreparedStatement pstmt;
       ResultSet rs;

       String url = context.getJobParameter("url","jdbc:mysql://xx.xx.xx.xx:3306/table");
       String driver = context.getJobParameter("driver","com.mysql.jdbc.Driver");
       String user = context.getJobParameter("user","user");
       String password = context.getJobParameter("password","password");

       userParasMap.put("url", url);
       userParasMap.put("driver", driver);
       userParasMap.put("user", user);
       userParasMap.put("password", password);

       connection = initConnection(userParasMap);
       String sql = "select `key`, `value` from udf_info";
       pstmt = connection.prepareStatement(sql);
       rs = pstmt.executeQuery();

       while (rs.next()) {
           remainedKey = rs.getString(1);
           remainedValue = rs.getString(2);
      }
  }

   public String eval(String... params) throws IOException {
       if (params != null && params.length != 0 && params.length % 2 <= 0) {
           HashMap<String, String> hashMap = new HashMap();
           for (int i = 0; i < params.length; i += 2) {
               hashMap.put(params[i], params[i + 1]);
               LOG.debug("now the key is " + params[i].toString() + "; now the value is " + params[i + 1].toString());
          }
           hashMap.put(remainedKey, remainedValue);
           ObjectMapper mapper = new ObjectMapper();
           String result = "{}";
           try {
               result = mapper.writeValueAsString(hashMap);
          } catch (Exception ex) {
               LOG.error("Get result failed." + ex.getMessage());
          }
           LOG.debug(result);
           return result;
      } else {
           return "{}";
      }
  }

   public static void main(String[] args) throws IOException {
       SimpleJsonBuild  sjb = new SimpleJsonBuild();
       System.out.println(sjb.eval("json1", "json2", "json3", "json4"));
  }
}

Add pipeline.global-job-parameters to Runtime Configuration on the Flink OpenSource SQL editing page. The format is as follows:

pipeline.global-job-parameters=url:'jdbc:mysql://x.x.x.x:xxxx/test',driver:com.mysql.jdbc.Driver,user:xxx,password:xxx

Flink OpenSource SQL

create function SimpleJsonBuild AS 'udf.SimpleJsonBuild';
create table dataGenSource(user_id string, amount int) with (
 'connector' = 'datagen',
 'rows-per-second' = '1', --Generate a piece of data per second.
 'fields.user_id.kind' = 'random', --Specify a random generator for the user_id field.
 'fields.user_id.length' = '3' --Limit the length of user_id to 3.
);
create table printSink(message STRING) with ('connector' = 'print');
insert into
printSink
SELECT
SimpleJsonBuild("name", user_id, "age", cast(amount as string))
from
dataGenSource;

Output

On the Flin Jobs page, locate your job, and click More > FlinkUI in the Operation column. On the displayed page, click Task Managers > Stdout to view the job output.

Click to enlarge

Parent topic: Functions

Previous topic: Type Inference

Next topic: Built-In Functions

Feedback

Was this page helpful?

Helpful Not helpful

Provide feedback

Thank you very much for your feedback. We will continue working to improve the documentation.See the reply and handling status in My Cloud VOC.

The system is busy. Please try again later.

Which of the following issues have you encountered?

Content is inconsistent with the product UI

Unclear descriptions

Lack of examples or code

Incorrect steps

Can't find what I need

Lack of best practices

Feedback (optional)

0/500

Select at least one type of issue, and enter your comments or suggestions.

Enter a maximum of 500 characters.

Submit Cancel

For any further questions, feel free to contact us through the chatbot.

Chatbot