如何使用睡觉接口在没有使用的情况下从Sink(或一般称为RichFunction)获取作业名称?
如标题所示。虽然getJobId
在RuntimeContext
中可用,但作业名称不可用。
https://nightlies.apache.org/flink/flink-docs-release-1.13/api/java/org/apache/flink/api/common/functions/RuntimeContext.html
尝试从配置中获取似乎效果不佳:
@Override
public void open(Configuration parameters) throws Exception {
String jobName = parameters.getString(PipelineOptions.NAME); // this is null
}
我们如何运行独立示例管道:
public static void main(String... args) {
try {
ParameterTool parameterTool = ParameterTool.fromArgs(args);
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// some pipeline setup
env.execute("This-is-job-name");
} catch (Exception e) {
// logging
}
解决方案
假设您将作业名称作为参数传递给作业,您希望将其设置如下:
public static void main(String... args) {
ParameterTool parameterTool = ParameterTool.fromArgs(args);
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().setGlobalJobParameters(parameters);
然后这应该会起作用
@Override
public void open(Configuration parameters) throws Exception {
ParameterTool params = (ParameterTool)
getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
String jobName = params.get(nameOfParameterWithJobName);
}
传递给open
的配置始终为空--这是一种不再使用的过时机制。未更改方法签名以避免破坏公共API。
将此类信息传递给RichFunction的另一个好方法是将其传递给构造函数。
相关文章