Dataflow中如何基于多个选项创建ValueProvider？多Value输入合并方案

首先明确官方限制：

NestedValueProvider 仅支持单个值输入，无法用于合并两个不同的取值。

以下是两种可落地的实现方案： #

方案1：自定义实现ValueProvider接口 #

适用于需要将合并后的值作为连接器参数、配置项等需要传入ValueProvider类型的场景：

• 自定义类实现ValueProvider接口，构造方法接收你需要合并的多个ValueProvider实例
• 重写get()方法，在其中编写你需要的合并逻辑（字符串拼接、数值计算、集合合并、参数组装等都可自定义）
• 重写isAccessible()方法，只有所有依赖的输入ValueProvider都可访问时才返回true，避免运行时取值错误

示例代码（Java版本，其他语言逻辑通用）：

public class CombinedValueProvider<T> implements ValueProvider<T> {
    // 可根据合并需求调整参数类型与数量，以下为字符串拼接示例
    private final ValueProvider<String> paramA;
    private final ValueProvider<String> paramB;
    private final String separator;

    public CombinedValueProvider(ValueProvider<String> paramA, ValueProvider<String> paramB, String separator) {
        this.paramA = paramA;
        this.paramB = paramB;
        this.separator = separator;
    }

    @Override
    public T get() {
        // 此处自定义合并逻辑，示例为拼接两个字符串参数
        return (T) (paramA.get() + separator + paramB.get());
    }

    @Override
    public boolean isAccessible() {
        return paramA.isAccessible() && paramB.isAccessible();
    }
}

使用时直接实例化该自定义类，传入你需要合并的多个ValueProvider即可。 #

方案2：在DoFn运行时逻辑中合并 #

如果你不需要在Pipeline构造阶段拿到合并后的值，仅需要在元素处理的运行时使用，可直接用更简单的实现方式：

• 自定义DoFn时，将需要合并的多个ValueProvider作为构造参数传入
• 在@ProcessElement方法中调用各个ValueProvider的get()方法，直接执行合并逻辑后使用

示例代码：

public class MyProcessFn extends DoFn<InputT, OutputT> {
    private final ValueProvider<String> config1;
    private final ValueProvider<Integer> config2;

    public MyProcessFn(ValueProvider<String> config1, ValueProvider<Integer> config2) {
        this.config1 = config1;
        this.config2 = config2;
    }

    @ProcessElement
    public void processElement(ProcessContext c) {
        // 运行时直接取值合并
        String combinedConfig = config1.get() + "_" + config2.get();
        // 后续处理逻辑
    }
}

注意事项 #

• 自定义ValueProvider时不要在构造阶段就调用get()方法，必须放在get()重写方法里，保证运行时才会解析运行时参数
• 如果合并逻辑复杂，可在自定义ValueProvider中添加更多辅助方法完成计算，只要保证对外暴露的接口符合ValueProvider规范即可

内容的提问来源于stack exchange，提问作者Георгий Новицкий