使用Flink时，字数始终在变化

Question

我正在尝试使用flink创建单词计数示例。这是单词数据的link（这是flink的github帐户的示例）

当我用简单的java程序计算单词数时：

public static void main(String[] args) throws Exception {
    int count = 0;
    for (String eachSentence : WordCountData.WORDS){
        String[] splittedSentence = eachSentence.toLowerCase().split("\\W+");
        for (String eachWord: splittedSentence){
            count++;
        }
    }
    System.out.println(count);
// result is 287
}

现在，当我使用flink进行此操作时，首先将句子拆分为单词。

DataStream<Tuple2<String, Integer>> readWordByWordStream = splitSentenceWordByWord(wordCountDataSource);

//...
public DataStream<Tuple2<String, Integer>> splitSentenceWordByWord(DataStream<String> wordDataSourceStream)
{
    DataStream<Tuple2<String, Integer>> wordByWordStream = wordDataSourceStream.flatMap(new TempTransformation());
    return wordByWordStream;
 }

• 这是我的TempTransformation类：

public class TempTransformation extends RichFlatMapFunction<String, Tuple2<String, Integer>> {

    @Override
    public void flatMap(String input, Collector<Tuple2<String, Integer>> collector) throws Exception
    {
        String[] splittedSentence = input.toLowerCase().split("\\W+");
        for (String eachWord : splittedSentence)
        {
            collector.collect(new Tuple2<String, Integer>(eachWord, 1));
        }
    }
}

• 现在，我将通过将其转换为KeyedStream（按单词键入）来计算单词数

    public SingleOutputStreamOperator<String> keyedStreamExample(DataStream<Tuple2<String, Integer>> wordByWordStream)
    {
        return wordByWordStream.keyBy(0).timeWindow(Time.milliseconds(1)).apply(new TempWindowFunction());
    }

• TempWindowFunction（）：

public class TempWindowFunction extends RichWindowFunction<Tuple2<String, Integer>, String, Tuple, TimeWindow> {
    private Logger logger = LoggerFactory.getLogger(TempWindowFunction.class);
    private int count = 0;
    @Override
    public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<String> out) throws Exception
    {
        logger.info("Key is:' {} ' and collected element for that key and count: {}", (Object) tuple.getField(0), count);
        StringBuilder builder = new StringBuilder();
        for (Tuple2 each : input)
        {
            String key = (String) each.getField(0);
            Integer value = (Integer) each.getField(1);
            String tupleStr = "[ " + key + " , " + value + "]";
            builder.append(tupleStr);
            count ++;
        }
        logger.info("All tuples {}", builder.toString());
        logger.info("Exit method");
        logger.info("----");
    }
}

• 在Flink的本地环境中运行此作业后，输出始终在变化，这是一些示例：

18:09:40,086 INFO  com.sampleFlinkProject.transformations.TempWindowFunction     - Key is:' rub ' and collected element for that key and count: 86
18:09:40,086 INFO  TempWindowFunction     - All tuples [ rub , 1]
18:09:40,086 INFO  TempWindowFunction     - Exit method
18:09:40,086 INFO  TempWindowFunction     - ----
18:09:40,086 INFO  TempWindowFunction     - Key is:' for ' and collected element for that key and count: 87
18:09:40,086 INFO  TempWindowFunction     - All tuples [ for , 1]
18:09:40,086 INFO  TempWindowFunction     - Exit method
18:09:40,086 INFO  TempWindowFunction     - ----

// another running outputs:

18:36:21,660 INFO  TempWindowFunction     - Key is:' for ' and collected element for that key and count: 103
18:36:21,660 INFO  TempWindowFunction     - All tuples [ for , 1]
18:36:21,660 INFO  TempWindowFunction     - Exit method
18:36:21,660 INFO  TempWindowFunction     - ----
18:36:21,662 INFO  TempWindowFunction     - Key is:' coil ' and collected element for that key and count: 104
18:36:21,662 INFO  TempWindowFunction     - All tuples [ coil , 1]
18:36:21,662 INFO  TempWindowFunction     - Exit method
18:36:21,662 INFO  TempWindowFunction     - ----

• 最后，这是执行设置

//...
final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.setParallelism(1);
//...

• 为什么Flink每次执行都会给出不同的输出？

Answer 1

应用程序中不确定性的一个来源是处理时间窗口（1 ms长）。每当您使用处理时间进行窗口化时，窗口最终都会包含在该时间间隔内显示并得到处理的所有事件。 （事件时间窗口的确具有确定性，因为它们是基于事件中的时间戳记的。）时间窗口太短会放大这种效果。