flinksql 导入数据到doris的时候怎么把sinklabel-prefix设置为动态的
可以使用 FlinkSQL 中的动态参数来设置 sink.label-prefix,具体步骤如下:
- 在 FlinkSQL 中定义动态参数:
SET execution.runtime-type=streaming;
SET sink.label-prefix=my_dynamic_prefix;
- 在创建 FlinkSQL 的 TableEnvironment 时,将动态参数作为参数传入:
TableEnvironment tEnv = TableEnvironment.create(
EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build());
tEnv.getConfig().getConfiguration().setString("execution.runtime-type", "${execution.runtime-type}");
tEnv.getConfig().getConfiguration().setString("sink.label-prefix", "${sink.label-prefix}");
- 在创建 FlinkSQL 的 Sink 时,从 TableEnvironment 中获取动态参数的值,然后将其设置为 Sink 的 label-prefix:
String labelPrefix = tEnv.getConfig().getConfiguration().getString("sink.label-prefix", "");
DorisSink dorisSink = new DorisSink(dorisConf);
dorisSink.setLabelPrefix(labelPrefix);
这样就可以动态地设置 sink.label-prefix 了。当需要改变 label-prefix 时,只需要修改动态参数即可,无需重新编译和部署代码。
原文地址: https://www.cveoy.top/t/topic/bQJR 著作权归作者所有。请勿转载和采集!