使用Flink时，字数总是在变化

Question

我正在尝试用flink创建字数示例。以下是单词数据的链接(这是来自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

应用程序中不确定性的一个来源是处理时间窗口(长度为1ms(。每当您使用处理时间进行窗口化时，窗口最终都会包含在该时间间隔内显示和处理的任何事件。(事件时间窗口的行为确实是决定性的，因为它们是基于事件中的时间戳。(窗口过短会夸大这种影响。