网站设计协议,银行门户网站是什么意思,广东汕头澄海手工外发加工网,提交百度收录前言
下图表示的是一个简单的flink-job的计算图#xff0c;这种图被称为DAG(有向无环图)#xff0c;表示的这个任务的计算逻辑#xff0c;无论是spark、hive、还是flink都会把用户的计算逻辑转换为这样的DAG#xff0c;数据的计算按照DAG触发#xff0c;理论上只要构建出…前言
下图表示的是一个简单的flink-job的计算图这种图被称为DAG(有向无环图)表示的这个任务的计算逻辑无论是spark、hive、还是flink都会把用户的计算逻辑转换为这样的DAG数据的计算按照DAG触发理论上只要构建出这样一个DAG图就可以描述清楚用户的计算逻辑在DAG的基础上将Node并行化就可以将整个job并行化。 在Flink之前的上一代流式计算框架Apache Storm的hello world如下节选了一部分从storm的helloworld代码可以很清楚的看到storm构建dag是依赖用户自己构建用户将自己脑中的dag图使用代码画出来line2创建了一个DAG的builderline4新增了一个节点line6也新增了一个节点dag画完了后在line16将DAG生成出来提交到集群执行。从这里可以看出storm构建DAG的逻辑是用户心中有图自己画出来。
// 实例化TopologyBuilder类。
TopologyBuilder topologyBuilder new TopologyBuilder();
// 设置喷发节点并分配并发数该并发数将会控制该对象在集群中的线程数。
topologyBuilder.setSpout(SimpleSpout, new SimpleSpout(), 1);
// 设置数据处理节点并分配并发数。指定该节点接收喷发节点的策略为随机方式。
topologyBuilder.setBolt(SimpleBolt, new SimpleBolt(), 3).shuffleGrouping(SimpleSpout);
Config config new Config();
config.setDebug(true);
if (args ! null args.length 0) {config.setNumWorkers(1);StormSubmitter.submitTopology(args[0], config, topologyBuilder.createTopology());
} else {// 这里是本地模式下运行的启动代码。config.setMaxTaskParallelism(1);LocalCluster cluster new LocalCluster();cluster.submitTopology(simple, config, topologyBuilder.createTopology());
}
再看一下flink的helloworld代码如下该代码对应的DAG就是文章开头的图片下面代码中line3获取一个执行的环境line6从9999端口读入数据line7做flatmapling15做分组操作line20对分组的数据做sum聚合line22执行任务通过和storm的helloworld的对比可以很明显的看出flink代码中很难看出DAG的样子flink专注的并不是用户去画DAG而是用户表达清楚自己的业务由flink将DAG画出并执行这也是flink会将storm慢慢淘汰的原因之一
public class Demo01_hello {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);env.socketTextStream(localhost, 9999).flatMap(new FlatMapFunctionString, Tuple2String, Integer() {Overridepublic void flatMap(String value, CollectorTuple2String, Integer out) throws Exception {for (String s : value.split( )) {out.collect(Tuple2.of(s, 1));}}}).keyBy(new KeySelectorTuple2String, Integer, String() {Overridepublic String getKey(Tuple2String, Integer value) throws Exception {return value.f0;}}).sum(1).print();env.execute();}
}
总结一下flink目前提供了多种api包裹flink-stream-apitable/sql-apipython-api这些api的表象不同但是底层都是将用户表达的逻辑翻译为DAG部署到集群上
那就从Hello-world开始吧
大数据的hello-word都是从wordcount开始的这是mapreduce时代的传承让我们再看一下flink的wordcount
public class Demo01_hello {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);env.socketTextStream(localhost, 9999).flatMap(new FlatMapFunctionString, Tuple2String, Integer() {Overridepublic void flatMap(String value, CollectorTuple2String, Integer out) throws Exception {for (String s : value.split( )) {out.collect(Tuple2.of(s, 1));}}}).keyBy(new KeySelectorTuple2String, Integer, String() {Overridepublic String getKey(Tuple2String, Integer value) throws Exception {return value.f0;}}).sum(1).print();env.execute();}
}
line3从StreamExecutionEnvironment获取了一个执行环境这个环境在本地就是local的在yarn上就是yarn的在k8s上就是k8s的
line4设置任务的并行度这里遇到了第一个概念并行度并行度表示任务的并行个数比如数据源kafka有2个分区那么最佳的并行度就是2因为一个分区只能被一个消费者消费并行度大于2则多余的消费者消费不到数据
line6设置了数据源为socket监听9999端口
line7对数据源的数据做flatmap操作输入是string输出是tuple2stringinteger
line15对tuple2stringinteger做了分组操作按照string分组这里涉及了另一个概念shuffleshuffle就是打乱的意思
line20对分组后的数据tuple2stringinteger做了sum操作计算出每一个string的数量
ling22执行任务
下图展示了该任务如何从代码变成可以运行的执行图运行在分布式环境中 可以看到上图中有四张图编写的代码会经历
StreamGraph - JobGraph - ExecutionGraph - 物理执行图最终提交到集群执行
1StreamGraph
aStreamNode表示每一个operator并且携带了这个operator的若干信息
bStreamEdge表示streamnode之间的边边上还携带了标识rebalance、hash、forward表示streamnode之间的数据传输方式
cStreamGraph其实已经很像前言中的dag图了但是还有些不同
2JobGraph
aJobVertexstreamgraph中的streamnode如果存在可以优化的情况比如operator-chain那么多个streamnode就可以合并为一个jobvertexoperator-chain的条件是streamedgeforward且前后两个streamnode并行度相同
b) IntermediateDatasetjobvertex的产出数据即若干个operator处理后的结果集
cJobEdge数据传输通道从intermediatedataset传输数据到下游jobvertex
3ExecutionGraph
aExecutionVertex: jobvertex的并行化节点
bExecutionJobVertexjobvertex对应的节点一一对应
cIntermediateResultPartition: 表示ExecutionVertex的输出结果一个ExecutionVertex对应一个IntermediateResultPartition
dIntermediateResult和IntermediateDataset一一对应
eExecutionEdge连接IntermediateResultPartition和ExecutionVertex一一对应
4物理执行图
aTask具体的调度task封装了operator的操作包括用户的逻辑
bResultPartition对应IntermediateResultPartition一一对应
cResultSubPartition是Resultpartition的子分区他的数量和下游的task有关如果source算子就一个所以他的ResultPartition就一个但是下游有两个flatmap算子所以这个ResultPartition会分成2个ResultSubPartition分别给下游两个flatmap算子消费
dInputChannel连接ResultSubPartition和下游task算子的数据通道
