特定于flink的解决方案可能是实现正在使用的flink操作符的自定义版本。我不相信这会把你带到一个好地方；只是分享一下想法以防有用。
关于如何实现自定义操作符的文档并不多，但是关于这个主题已经有了深入的讨论。

首先，我要在aws上提交一个特性请求来支持您的用例。那将是最干净的解决办法。
第二，我不想找到任何方法来覆盖 open() . 由于您所处的环境没有太多的控制权，我认为这些方法要么根本不起作用，要么很脆弱，随着环境的更新而中断。
我会在各自的udf方法中进行延迟初始化，当然也会在一些常用的实用程序方法中考虑到这一点。

private Counter counter;

@Override
public Integer map(String value) {
    if (counter == null) {
        RuntimeContext ctx = getRuntimeContext();
        counter = ctx.getMetricGroup().counter("outputs");
    }
    counter.inc();
    return Integer.parseInt(value);
}

使用byte-buddy，您可以创建 Package 器或java代理，两者都可以实现这一目标。如果您在 Package 类的构造函数调用方面遇到困难，那么使用byte-buddy会出现同样的问题，因为任何库都绑定到jvm给定的约束。
要创建java代理，请使用 AgentBuilder . 然后可以使用 type 步骤，例如实现某个接口或扩展某个类的所有类型。为了 transform ，byte buddy提供了一个名为 Advice ，它允许您添加其他代码，例如：

class MyAdvice {
  @Advice.OnMethodEnter
  static void enter() { System.out.println("Hello"); }
}

通过

builder = builder.visit(Advice.to(MyAdvice.class).on(named("foo")));

例如，可以在指定类型的所有名为“foo”的方法的开头打印helloworld。您可以在的包文档中找到有关java代理的更多信息 java.instrument 包裹。

原始asker的回答：我们通过为streamexecutionenvironment创建一个bytebuddy代理来解决这个问题，该代理拦截对getstreamgraph的调用，并将每个节点的jobvertexpress（使用反射）重铸到一个扩展了原始类类型但包含了我们的自定义逻辑的类中。因为不同的类需要不同的参数，所以我们使用objenesis示例化代理而不调用构造函数。为了解决通常在构造函数中设置的私有字段为空的问题，我们使用反射来改变所有私有字段的可见性，然后将每个字段值从原始对象复制到代理对象。
我们没有追求rafael winterhalter提出的代理解决方案，因为它要求能够在每个工作机示例上运行代理设置代码，这类似于希望在每个工作机上启动metrics代理的原始问题。虽然我在最初的问题中没有说明这一点，但是创建代理对象的代码发生在flink作业管理机器上，而不是工作机器上。