Re: How can I set job parameter in flink sql

[Qingsheng Ren]

Hi,

You can take use of the configuration “pipeline.global-job-parameters” [1] to 
pass your custom configs all the way into the UDF. For example you can execute 
this in SQL client:

SET pipeline.global-job-parameters=black_list_path:/root/list.properties;

Then you can get the value “/root/list.properties” by 
context.getJobParameter(“black_list_path”, “your_default_value”) in the open() 
method of your UDF.

[1] 
https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/config/#pipeline-global-job-parameters

Cheers, 

Qingsheng

> On May 11, 2022, at 14:36, wang <24248...@163.com> wrote:
> 
> Hi dear engineer,
> 
> I want to override the function open() in my UDF, like:
> 
> 
> 
> In open() function, I want to fetch the configred value "black_list_path", 
> then simply print that value out. And I config this value in ./sql-client.sh 
> console:
> 
> SET black_list_path = /root/list.properties
> 
> Then I run this UDF, but what printed is /config/list.properties(this is the 
> default value as I set in context.getJobParameter("black_list_path", 
> "/config/list/properties")), not /root/list.properties which I set in 
> ./sql-client.sh console.
> 
> So could you please show me the correct way to set black_list_path is sql ? 
> Thanks so much!
> 
> 
> Thanks && Reards,
> Hunk
> 
> 
> 
> 
> 
>  

Re:Re: How can I set job parameter in flink sql

[wang]

Ok, got it. Thanks so much！

Regards,
Hunk



--
发自我的网易邮箱手机智能版



在 2022-05-11 16:46:14，yuxia  写道：

Hi, AFAK,  you can't get the parameter setted via Flink SQL client in udf.


If you still want to get the parameters in your udf, you can use the following 
code to set the parameter:


env = StreamExecutionEnvironment.getExecutionEnvironment
parameter = new HashMap();
parameter.put("black_list_path", "")
env.getConfig.setGlobalJobParameters(Configuration.fromMap(m))


Then, you can get the parameter using 
context.getJobParameter("black_list_path", "/config/list.properties"); in udf.


Best regards,
Yuxia


发件人: "wang" <24248...@163.com>
收件人: "User" , "user-zh" 
发送时间: 星期三, 2022年 5 月 11日 下午 2:44:20
主题: How can I set job parameter in flink sql



Hi dear engineer,


I want to override the function open() in my UDF, like:




|
public class BlackListConvertFunction extends ScalarFunction {


@Override
public void open(FunctionContext context) throws Exception {
String path = context.getJobParameter("black_list_path", 
"/config/list.properties");
System.out.println(path);
}


public Double eval(String scores) {
// some logics
return 0.0;
}
}

|






In open() function, I want to fetch the configred value "black_list_path", then 
simply print that value out. And I config this value in ./sql-client.sh console:


SET black_list_path = /root/list.properties


Then I run this UDF, but what printed is /config/list.properties(this is the 
default value as I set in context.getJobParameter("black_list_path", 
"/config/list/properties")), not /root/list.properties which I set in 
./sql-client.sh console.


So could you please show me the correct way to set black_list_path is sql ? 
Thanks so much!




Thanks && Reards,
Hunk






 

Re: How can I set job parameter in flink sql

[yuxia]

Hi, AFAK, you can't get the parameter setted via Flink SQL client in udf. 

If you still want to get the parameters in your udf, you can use the following 
code to set the parameter: 

env = StreamExecutionEnvironment.getExecutionEnvironment 
parameter = new HashMap(); 
parameter .put(" black_list_path ", "") 
env.getConfig.setGlobalJobParameters(Configuration.fromMap(m)) 

Then, you can get the parameter using 
context.getJobParameter("black_list_path", "/config/list.properties"); in udf. 

Best regards, 
Yuxia 


发件人: "wang" <24248...@163.com> 
收件人: "User" , "user-zh"  
发送时间: 星期三, 2022年 5 月 11日 下午 2:44:20 
主题: How can I set job parameter in flink sql 

Hi dear engineer, 

I want to override the function open() in my UDF, like: 


public class BlackListConvertFunction extends ScalarFunction { 

@Override 
public void open(FunctionContext context) throws Exception { 
String path = context.getJobParameter("black_list_path", 
"/config/list.properties"); 
System.out.println(path); 
} 

public Double eval(String scores) { 
// some logics 
return 0.0; 
} 
} 





In open() function, I want to fetch the configred value "black_list_path", then 
simply print that value out. And I config this value in ./sql-client.sh 
console: 

SET black_list_path = /root/list.properties 

Then I run this UDF, but what printed is /config/list.properties (this is the 
default value as I set in context.getJobParameter("black_list_path", " 
/config/list/properties ")) , not /root/list.properties which I set in 
./sql-client.sh console. 

So could you please show me the correct way to set black_list_path is sql ? 
Thanks so much! 


Thanks && Reards, 
Hunk 

How can I set job parameter in flink sql

[wang]

Hi dear engineer,


I want to override the function open() in my UDF, like:




|
public class BlackListConvertFunction extends ScalarFunction {


@Override
public void open(FunctionContext context) throws Exception {
String path = context.getJobParameter("black_list_path", 
"/config/list.properties");
System.out.println(path);
}


public Double eval(String scores) {
// some logics
return 0.0;
}
}


|






In open() function, I want to fetch the configred value "black_list_path", then 
simply print that value out. And I config this value in ./sql-client.sh console:


SET black_list_path = /root/list.properties


Then I run this UDF, but what printed is /config/list.properties(this is the 
default value as I set in context.getJobParameter("black_list_path", 
"/config/list/properties")), not /root/list.properties which I set in 
./sql-client.sh console.


So could you please show me the correct way to set black_list_path is sql ? 
Thanks so much!




Thanks && Reards,
Hunk

How can I set job parameter in flink sql

[wang]

Hi dear engineer,


I want to override the function open() in my UDF, like:




In open() function, I want to fetch the configred value "black_list_path", then 
simply print that value out. And I config this value in ./sql-client.sh console:


SET black_list_path = /root/list.properties


Then I run this UDF, but what printed is /config/list.properties(this is the 
default value as I set in context.getJobParameter("black_list_path", 
"/config/list/properties")), not /root/list.properties which I set in 
./sql-client.sh console.


So could you please show me the correct way to set black_list_path is sql ? 
Thanks so much!




Thanks && Reards,
Hunk