Hi~朋友,关注置顶防止错过消息
什么是流处理?
流处理就是我们对流动的数据(无限的数据)进行处理,通常我们会提前设置好算子(也就是你的处理逻辑),当数据到达后对数据进行处理。
流处理的模型什么?
为了表达我们复杂的计算逻辑,Flink使用DAG图来表达整个计算逻辑,DAG的每一个点都代表一个基本的逻辑计算单元(算子),数据会按照DAG图的边进行流动,从数据源出发, 流经算子,最后通过Sink节点将结果输出到外部系统。
如上图所示,DAG图只是简单的逻辑图,不包含并发(也就是实际的执行情况),在实际执行的时候,每个算子可能被分配到多个实例上,对于同一个实例的上下游算子可以不需要网络, 但是如果上下游算子不在同一个实例上则需要通过网络进行数据传输,如下图:
Tuple
Tuple在Flink中我们经常使用的一个类,Tuple用来存放固定个数的属性,最多支持存放25个属性,
从上图可以看出,Tuple有很多个实现类,Tuple2代表可以存放2个属性,Tuple25代表可以存放25个属性。
FlatMap
FlatMap的作用是输入一个元素,输出多个元素,DataStream.flatMap方法通常需要传入一个FlatMapFunction,该函数中有一个flatMap方法用于将一个元素转换为多个元素,
public interface FlatMapFunction<T, O> extends Function, Serializable {
void flatMap(T value, Collector<O> out) throws Exception;
}
从类图上我们可以看到我们FlatMapFunction继承自Function,同时他有一个子类抽象类RichFlatMapFunction,RichFlatMapFunction又继承自AbstractRichFunction,AbstractRichFunction 又实现了RickFunction,RickFunction最终又继承了Function。
Function
Function是Flink中我们所有自定义操作的接口(包括Flink自己实现的一些操作),该接口没有任何方法,只是用来声明实现该接口的类可以用做算子的处理逻辑。
public interface Function extends java.io.Serializable {}
RickFunction
RickFunction在Function的声明外,加入了一些方法,如下:
@Public
public interface RichFunction extends Function {
void open(Configuration parameters) throws Exception;
void close() throws Exception;
RuntimeContext getRuntimeContext();
IterationRuntimeContext getIterationRuntimeContext();
void setRuntimeContext(RuntimeContext t);
}
WordCount
我们的WordCount程序主要分为4个步骤:
关于运算逻辑:
public class WordCount {
public static void main(String[] args) throws Exception {
// 步骤1
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 步骤2
DataStreamSource<String> source = env.fromElements(WORDS);
// 步骤3
source.flatMap(new Tokenizer()).keyBy(new KeySelector<Tuple2<String, Integer>, Object>() {
@Override
public Object getKey(Tuple2<String, Integer> value) throws Exception {
return value.f0;
}
}).sum(1).addSink(new SinkFunction<Tuple2<String, Integer>>() {
@Override
public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
System.out.println("单词:" + value.f0 + ",数量:" + value.f1);
}
});
// 步骤4
env.execute("WordCount Stream");
}
private static class Tokenizer extends RichFlatMapFunction<String, Tuple2<String, Integer>> {
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
String[] tokens = value.toLowerCase().split("\\W+");
for (String token : tokens) {
if (token.length() > 0) {
out.collect(new Tuple2<>(token, 1));
}
}
}
}
public static final String[] WORDS =
new String[] {
"To be, or not to be,--that is the question:--",
"Whether 'tis nobler in the mind to suffer",
"The slings and arrows of outrageous fortune",
"Or to take arms against a sea of troubles,",
"And by opposing end them?--To die,--to sleep,--",
"No more; and by a sleep to say we end",
"The heartache, and the thousand natural shocks",
"That flesh is heir to,--'tis a consummation",
"Devoutly to be wish'd. To die,--to sleep;--",
"To sleep! perchance to dream:--ay, there's the rub;",
"For in that sleep of death what dreams may come,",
"When we have shuffled off this mortal coil,",
"Must give us pause: there's the respect",
"That makes calamity of so long life;",
"For who would bear the whips and scorns of time,",
"The oppressor's wrong, the proud man's contumely,",
"The pangs of despis'd love, the law's delay,",
"The insolence of office, and the spurns",
"That patient merit of the unworthy takes,",
"When he himself might his quietus make",
"With a bare bodkin? who would these fardels bear,",
"To grunt and sweat under a weary life,",
"But that the dread of something after death,--",
"The undiscover'd country, from whose bourn",
"No traveller returns,--puzzles the will,",
"And makes us rather bear those ills we have",
"Than fly to others that we know not of?",
"Thus conscience does make cowards of us all;",
"And thus the native hue of resolution",
"Is sicklied o'er with the pale cast of thought;",
"And enterprises of great pith and moment,",
"With this regard, their currents turn awry,",
"And lose the name of action.--Soft you now!",
"The fair Ophelia!--Nymph, in thy orisons",
"Be all my sins remember'd."
};
}
DataStream
从上面的使用可以看出,Flink DataStream AP整个的核心就是代表流数据的DataStream对象,我们整个逻辑运算都是围绕DataStream对象进行操作然后产生新的DataStream对象,对于DataStream 单条记录我们可以进行filter、map等操作,或者基于window对多条记录进行操作,同时我们也可以将单条流(DataStream)进行拆分,也可以对多条流进行合并,如下图:
在Flink中,最基础的流是DataStream,但是经过上面的操作以后可能会产生各种各样的流类型,目前Flink中的流的转换关系图如下:
在上面各式各样的流中,每个流都有自己独特的特点及操作,同时多种流之间也支持某些通用的操作,关于每种流的使用我们下篇进行详细实践。