Flink使用(一) Streaming API处理有界流

前提：

目前，Flink版本支持，批流处理使用一套API完成。

即，使用DataStream API既能处理流数据，又能处理批数据（有界流）。

如何使用DataStream API处理数据，并且以BATCH的方式运行呢？

方式一：在程序中指定运行模式（不推荐，调试可以）：

        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

方式二：

提交任务时，使用bin/flink/run -Dexecution.runtime-mode=BATCH 指定BATCH方式运行程序。

使用BATCH方式运行的目的？

我理解是将最终的结果进行输出，而不是获取一条记录后就输出一条结果。

示范例子

例子数据源word.txt,只有三行数据

hello word
hello flink
hello java

正常情况下，使用DataStream API对数据进行流式处理。

package org.galaxy.foundation.core.courses;import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;public class Stream10 {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource lineDataStreamSource = env.readTextFile("/tmp/foundation-core/words.txt");SingleOutputStreamOperator> wordAndOneTuple=  lineDataStreamSource.flatMap((String line, Collector> out) -> {String[] words = line.split(" ");for (String word : words) {out.collect(Tuple2.of(word, 1L));}}).returns(Types.TUPLE(Types.STRING,Types.LONG));KeyedStream, String> wordAndOneKeyedStream = wordAndOneTuple.keyBy(data -> data.f0);SingleOutputStreamOperator> sum = wordAndOneKeyedStream.sum(1);sum.print();env.execute();}}

打印结果：

3> (hello,1)
2> (java,1)
7> (flink,1)
6> (word,1)
3> (hello,2)
3> (hello,3)

可以看出，每来一条记录，即产生一条输出。

重点来了，使用DataStream API，以BATCH方式处理数据

只要在env初始化的下面增加一行

env.setRuntimeMode(RuntimeExecutionMode.BATCH);

重新运行，打印结果：

6> (word,1)
2> (java,1)
7> (flink,1)
3> (hello,3)

可以看出，将最终的汇总结果，一次完整的打印出来。

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！