Flink（1.3.2）向每个运营商播放一次记录

我有一个像这样的Executiongraph：

{"nodes":[{"id":1,"type":"Source: AggregatedData","pact":"Data Source","contents":"Source: AggregatedData","parallelism":1},{"id":2,"type":"AddVirtualKeyFunction","pact":"Operator","contents":"AddVirtualKeyFunction","parallelism":4,"predecessors":[{"id":1,"ship_strategy":"REBALANCE","side":"second"}]},{"id":3,"type":"Source: FilterInformation","pact":"Data Source","contents":"Source: FilterInformation","parallelism":1},{"id":4,"type":"BroadcastFilterInformation","pact":"Operator","contents":"BroadcastFilterInformation","parallelism":1,"predecessors":[{"id":3,"ship_strategy":"FORWARD","side":"second"}]},{"id":7,"type":"ConnectAndApplyFilterFunction","pact":"Operator","contents":"ConnectAndApplyFilterFunction","parallelism":4,"predecessors":[{"id":2,"ship_strategy":"HASH","side":"second"},{"id":4,"ship_strategy":"HASH","side":"second"}]},{"id":8,"type":"Sink: OutputFilteredData","pact":"Data Sink","contents":"Sink: OutputFilteredData","parallelism":4,"predecessors":[{"id":7,"ship_strategy":"FORWARD","side":"second"}]}]}

（可以在这里看到：https://flink.apache.org/visualizer/）

我有一个聚合数据流（“AggregatedData”，ID = 1），需要通过来自另一个流的一些过滤器（“FilterInformation”，ID = 3）进行过滤 .

我最初做的是在我的“ConnectAndApplyFilterFunction”（ID = 7）中使用运算符状态，这在技术上工作正常，但仅限于1的并行性 .

目前，我正在做一些黑客攻击：在"AddVirtualKeyFunction"中，我将聚合数据映射到 Tuple2<Integer, AggregatedData> ，其中整数（f0）是从0到19的随机生成的数字：

@Override
public Tuple2<Integer, AggregatedData> map(AggregatedData value) throws Exception {
    return new Tuple2<>(ThreadLocalRandom.current().nextInt(this.virtualKeyCount), value);
}

"BroadcastFilterInformation"是一个flatMap，每次收到一个新的FilterInformation时都会发布 Tuple2<Integer, FilterInfo> 20 Times（f0 0-19）：

@Override
public void flatMap(FilterInfo filterInfo, Collector<Tuple2<Integer, FilterInfo>> collector) throws Exception {
    if (this.currentLatestTimestamp < filterInfo.getLastUpdateTime()) {
        this.currentLatestTimestamp = filterInfo.getLastUpdateTime();

        for (int i = 0; i < this.broadcastCount; i++) {
            collector.collect(new Tuple2<>(i, filterInfo));
        }
    }
}

我现在连接两个流并通过"virtual key"（ Tuple2.f0 ）键入它们 . 我在"ConnectAndapplyFilterFunction"（ID = 7）中将我的 FilterInfo 的20份副本保存在键控状态 .

工作正常，我可以在我的主要道路上使用并行性 . 但为什么我使用20“虚拟键”而我的并行度只有4？因为只有4个键，所以不会使用所有运算符（2个运算符在我的测试中没有接收到任何数据） .

有什么办法可以从一个流中广播一些数据，以便另一端的每个操作员都能收到它自己的副本吗？