我有两条小溪。第一个是基于时间的流，我使用 countTimeWindow 接收前10个数据点，计算统计值。我手动使用了变量 cnt 只保留第一个窗口并过滤剩余值，如下面的代码所示。
然后，我想用这个值来过滤主流，以便得到大于我在窗口流中计算的stat值的值。
但是，我不知道如何合并或计算这两个流来实现我的目标。
我的场景是，如果我把第一个stat值转换成广播变量，然后我把它交给主流，这样我就可以根据广播变量中的stat值过滤传入的值。
下面是我的代码。

import com.sun.org.apache.xpath.internal.operations.Bool;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.functions.windowing.*;
import org.apache.flink.util.Collector;
import scala.Int;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;

public class ReadFromKafka {
    static int cnt = 0;
    public static void main(String[] args) throws Exception{
        // create execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "flink");

        DataStream<String> stream = env
                .addSource(new FlinkKafkaConsumer09<>("flinkStreaming11", new SimpleStringSchema(), properties));

        env.enableCheckpointing(1000);

//Time based window stream
        DataStream<String> process = stream.countWindowAll(10).process(new ProcessAllWindowFunction<String, Tuple2<Double, Integer>, GlobalWindow>() {
            @Override
            public void process(Context context, Iterable<String> iterable, Collector<Tuple2<Double, Integer>> collector) throws Exception {
                Double sum = 0.0;

                int n = 0;
                List<Double> listDouble = new ArrayList<>();
                for (String in : iterable) {
                    n++;
                    double d = Double.parseDouble(in);
                    sum += d;
                    listDouble.add(d);
                }

                cnt++;
                Double[] sd = listDouble.toArray(new Double[listDouble.size()]);
                double mean = sum / n;

                double sdev = 0;
                for (int i = 0; i < sd.length; ++i) {
                    sdev += ((sd[i] - mean) * (sd[i] - mean)) / (sd.length - 1);
                }
                double standardDeviation = Math.sqrt(sdev);
                collector.collect(new Tuple2<Double, Integer>(mean + 3 * standardDeviation, cnt));
            }
        }).filter(new FilterFunction<Tuple2<Double, Integer>>() {
            @Override
            public boolean filter(Tuple2<Double, Integer> doubleIntegerTuple2) throws Exception {
                Integer i1 = doubleIntegerTuple2.f1;
                if (i1 > 1)
                    return false;
                else
                    return true;
            }
        }).map(new RichMapFunction<Tuple2<Double, Integer>, String>() {
            @Override
            public String map(Tuple2<Double, Integer> doubleIntegerTuple2) throws Exception {
                return String.valueOf(doubleIntegerTuple2.f0);
            }
        });

//I don't think that this is not a proper solution.
        process.union(stream).filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                return false;
            }
        })

        env.execute("InfluxDB Sink Example");

        env.execute();
    }
}