thanks，Yang.
Btw, 恭喜晋升apache flink的PMC！
Best，
tanjialiang.
在2022年5月12日 10:32，Yang Wang 写道：
可以临时通过-D "$internal.pipeline.job-id="来自定义job id，但是个内部参数
你可以看下[1]，了解更多讨论的信息
[1]. https://issues.apache.org/jira/browse/FLINK-19358
Best,
Yang
谭家良 于2022年5月11日周三 22:17写道：
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 =
| |
Jerry Guo
|
|
wangyixuhongm...@163.com
|
转发的原邮件
| 发件人 | 徐战辉 |
| 发送日期 | 2022年5月12日 10:38 |
| 收件人 | user-zh@flink.apache.org |
| 主题 | 基于flink sql作业失败与取消状况下的作业参数设置咨询(flink-1.14.2) |
hello, 请教下，如何设置flink配置及作业参数，在取消作业重新部署、flink作业失败重跑情况下，保证不丢失数据。
目前有一份作业，开启checkpoint, cancel 后重新启动，发现数据会丢失1小部分。
1. flink.conf
execution.checkpointing.interval: 1
execution.checkpointing.externalized-checkpoint-retention:
RETAIN_ON_CANCELLATION
可以临时通过-D "$internal.pipeline.job-id="来自定义job id，但是个内部参数
你可以看下[1]，了解更多讨论的信息
[1]. https://issues.apache.org/jira/browse/FLINK-19358
Best,
Yang
谭家良 于2022年5月11日周三 22:17写道：
>
>
> 我使用的Application模式：Kubernetes
> 我使用的HA模式：Kubernetes HA
>
>
> 目前Application + HA发现所有的Job
我使用的Application模式：Kubernetes
我使用的HA模式：Kubernetes HA
目前Application + HA发现所有的Job ID都变成了“”，但是在HistoryServer中，他会使用Job
ID作为归档文件的文件名。那么如果这样，HistoryServer应该怎么与Application + HA相结合呢？
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("
不好意思，我尝试复现你的问题，但是我没有 hbase 环境，不过看起来是只有当 STORED BY
'org.apache.hadoop.hive.hbase.HBaseStorageHandler' 有问题？
我之后空了再debug 看看。
不过我看了一下 flink 这块的代码，从 flink 这块的代码来看，应该是 get 这个 hive 表之后，它的 StorageDescriptor 的
inputformat 为 null，然后 Class.forName(inputformat) 就报错 NPE了。
应该是这块代码有点问题。
如果你方便的话，可以辛苦帮忙建一个
Really excited to see the very first release of the flink-table-store!
Kudos to everyone who helped with this effort!
Cheers,
Jiangjie (Becket) Qin
On Wed, May 11, 2022 at 1:55 PM Jingsong Lee
wrote:
> The Apache Flink community is very happy to announce the release of Apache
> Flink Table
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",
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
11 matches
Mail list logo
