Flink与Spark Streaming谁更厉害

运行模型

Spark Streaming 是微批处理，运行的时候需要指定批处理的时间，每次运行 job 时处理一个批次的数据

Flink 是基于事件驱动的，事件可以理解为消息。事件驱动的应用程序是一种状态应用程序，它会从一个或者多个流中注入事件，通过触发计算更新状态，或外部动作对注入的事件作出反应。

运行角色

Spark Streaming 运行时的角色(standalone 模式)主要有：

Master:主要负责整体集群资源的管理和应用程序调度;

Worker:负责单个节点的资源管理，driver 和 executor 的启动等;

Driver:用户入口程序执行的地方，即 SparkContext 执行的地方，主要是 DAG 生成、stage 划分、task 生成及调度;

Executor:负责执行 task，反馈执行状态和执行结果。

Flink 运行时的角色(standalone 模式)主要有:

Jobmanager: 协调分布式执行，他们调度任务、协调 checkpoints、协调故障恢复等。至少有一个 JobManager。高可用情况下可以启动多个 JobManager，其中一个选举为 leader，其余为 standby;

Taskmanager: 负责执行具体的 tasks、缓存、交换数据流，至少有一个 TaskManager;

Slot: 每个 task slot 代表 TaskManager 的一个固定部分资源，Slot 的个数代表着 taskmanager 可并行执行的 task 数。

编程模型对比

编程模型对比，主要是对比 flink 和 Spark Streaming 两者在代码编写上的区别。

Spark Streaming

Spark Streaming 与 kafka 的结合主要是两种模型：

基于 receiver dstream;

基于 direct dstream。

以上两种模型编程机构近似，只是在 api 和内部数据获取有些区别，新版本的已经取消了基于 receiver 这种模式，企业中通常采用基于 direct Dstream 的模式。

val Array(brokers, topics) = args// 创建一个批处理时间是2s的context  

 val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount")  

 val ssc = new StreamingContext(sparkConf, Seconds(2))  

 // 使用broker和topic创建DirectStream  

 val topicsSet = topics.split(",").toSet  

 val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)  

 val messages = KafkaUtils.createDirectStream[String, String]( ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topicsSet, kafkaParams))  

 // Get the lines, split them into words, count the words and print  

 val lines = messages.map(_.value)  

 val words = lines.flatMap(_.split(" "))  

 val wordCounts = words.map(x => (x, 1L)).reduceByKey(_ + _)  

 wordCounts.print() // 启动流  

 ssc.start()  

 ssc.awaitTermination() 

通过以上代码我们可以 get 到：

设置批处理时间

创建数据流

编写transform

编写action

启动执行

Flink

接下来看 flink 与 kafka 结合是如何编写代码的。Flink 与 kafka 结合是事件驱动，大家可能对此会有疑问，消费 kafka 的数据调用 poll 的时候是批量获取数据的(可以设置批处理大小和超时时间)，这就不能叫做事件触发了。而实际上，flink 内部对 poll 出来的数据进行了整理，然后逐条 emit，形成了事件触发的机制。 下面的代码是 flink 整合 kafka 作为 data source 和 data sink：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 

 env.getConfig().disableSysoutLogging(); 

 env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000)); 

 env.enableCheckpointing(5000); // create a checkpoint every 5 seconds 

 env.getConfig().setGlobalJobParameters(parameterTool); // make parameters available in the web interface 

 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);  

 // ExecutionConfig.GlobalJobParameters 

 env.getConfig().setGlobalJobParameters(null); DataStream<KafkaEvent> input = env 

 .addSource( new FlinkKafkaConsumer010<>( 

 parameterTool.getRequired("input-topic"), new KafkaEventSchema(), 

 parameterTool.getProperties()) 

 .assignTimestampsAndWatermarks(new CustomWatermarkExtractor())).setParallelism(1).rebalance() 

 .keyBy("word") 

 .map(new RollingAdditionMapper()).setParallelism(0); 

  

 input.addSink( new FlinkKafkaProducer010<>( 

 parameterTool.getRequired("output-topic"), new KafkaEventSchema(), 

 parameterTool.getProperties())); 

  

 env.execute("Kafka 0.10 Example"); 

从 Flink 与 kafka 结合的代码可以 get 到：

注册数据 source

编写运行逻辑

注册数据 sink

调用 env.execute 相比于 Spark Streaming 少了设置批处理时间，还有一个显著的区别是 flink 的所有算子都是 lazy 形式的，调用 env.execute 会构建 jobgraph。client 端负责 Jobgraph 生成并提交它到集群运行;而 Spark Streaming的操作算子分 action 和 transform，其中仅有 transform 是 lazy 形式，而且 DAG 生成、stage 划分、任务调度是在 driver 端进行的，在 client 模式下 driver 运行于客户端处。

任务调度原理

Spark 任务调度

Spark Streaming 任务如上文提到的是基于微批处理的，实际上每个批次都是一个 Spark Core 的任务。对于编码完成的 Spark Core 任务在生成到最终执行结束主要包括以下几个部分：

构建 DAG 图;

划分 stage;

生成 taskset;

调度 task。

具体可参考图 5：

关于流式计算：Flink与Spark Streaming谁的拳头更硬

对于 job 的调度执行有 fifo 和 fair 两种模式，Task 是根据数据本地性调度执行的。 假设每个 Spark Streaming 任务消费的 kafka topic 有四个分区，中间有一个 transform操作(如 map)和一个 reduce 操作，如图 6 所示：

关于流式计算：Flink与Spark Streaming谁的拳头更硬

假设有两个 executor，其中每个 executor 三个核，那么每个批次相应的 task 运行位置是固定的吗?是否能预测? 由于数据本地性和调度不确定性，每个批次对应 kafka 分区生成的 task 运行位置并不是固定的。

Flink 任务调度

对于 flink 的流任务客户端首先会生成 StreamGraph，接着生成 JobGraph，然后将 jobGraph 提交给 Jobmanager 由它完成 jobGraph 到 ExecutionGraph 的转变，最后由 jobManager 调度执行。