你问的是flink的任务链具体怎么操作是吧？帮你详细解答下。

       Flink 中的每个算子都可以设置并行度，每个算子的一个并行度实例就是一个 subTask。由于 Flink 的 TaskManager 运行 Task 的时候是每个 Task 采用一个单独的线程，这会带来很多线程切换和数据交换的开销，进而影响吞吐量。
       为了避免数据在网络或线程之间传输导致的开销，Flink 会在 JobGraph 阶段，将代码中可以优化的算子优化成一个算子链（Operator Chains）以放到一个 Task 中执行。
用户也可以自己指定相应的链条，将相关性非常强的转换操作绑定在一起，这样能够让转换过程中上下游的 Task 在同一个 Pipeline 中执行，进而避免因为数据在网络或者线程间传输导致的开销，提高整体的吞吐量和延迟。
       一般情况下，Flink 在 Map 操作中默认开启 TaskChain，以提高 Flink 作业的整体性能。
如图1，Source 和 Map 在优化后，组成一个算子链，作为一个 task 运行在一个线程上，其简图如 Condensed view 所示，并行图如 parellelized view 所示。

Flink提供了更细粒度的任务链控制方法，用户可根据需求创建任务链或禁止任务链。

禁用全局任务链

evn.disableOperatorChaining();1

关闭全局任务链后，创建对应Operator Chain，需要用户先指定操作符，然后再调用startNewChain()方法创建。

dataStream.keyBy(0).filter().map().startNewChain().map();1

   startNewChain方法创建的链条只对调用方法的前一个操作符和后一个操作符有效，不影响其他的。比如示例中新建的链条只有map->map，对前面的filter无效。
       禁用全局任务链会影响整体任务执行的情况，禁用前，要清楚任务执行的流程，否则可能造成非预期的结果。

禁用局部任务链

       如果不想关闭整体算子上的链条，只是想关闭部分算子上链条绑定，可以使用disableChaining()方法禁用当前操作符上的链条。

dataStream.keyBy(0).filter().map().disableChaining()1

上述代码只会禁用map操作上的任务链，不会影响其他操作符。

flink 中并行任务的分配

• Flink 中每一个 TaskManager 都是一个JVM进程，它可能会在独立的线程上执行一个或多个 subtask

• 为了控制一个 TaskManager 能接收多少个 task， TaskManager 通过 task slot 来进行控制（一个 TaskManager 至少有一个 slot）

slot 主要隔离内存，cpu 是slot之间共享的。也就是说4核的机器 ，内存足够，可以把slot设置为8。最多能同时运行8个任务。建议一个核心数分配一个slot

这种图中 source、map 合成的task的并行度为6
keyby 、window、apply合成的task的并行度为6
sink的并行度为1
总共有13个task
但是不是需要13个slot才能满足这个并行度的要求

不同的算子操作复杂度不同
我们可以称像source map sink 这种 计算不复杂的算子称为非资源密集型的算子 aggregate reduce sum window 这种计算复杂的算子称为为资源密集型的算子

如果把这两种算子的优先级看作相同，平等的分配到slo中，当数据流source 来的数据速率相同时，会造成有些slot一直在跑复杂的算子，一直在运行中，当时一直跑简单算子的slot就会很空闲。

flink 这里是 非资源密集型的 算子和资源密集型的算子可以分配到同一个slot中 ，这样所有的slot之间任务就会平等，不会存在一直空闲一直高负载。

一个task的并行度是6 就会分为6个并行的task来跑，这六个task不能分配到同一个slot中必须一个slot只有一个。 也就是说 当你的集群的slot只有6 ，你不能设置算子的 并行度超过6。

flink 也能做到把非资源密集型和资源密集型的算子分到不同的slot中 这里需要设置共享组，非资源 密集型 的算子在一个共享组，资源密集 型的算子在一个共享组，这样这两种算子就不会共享的使用slot。默认情况下算有算子都属于同一个共享组，共享所有slot。

• 默认情况下，Flink 允许子任务共享 slot，即使它们是不同任务的子任务但是可以分配到同一个slot上。 这样的结果是，一个 slot 可以保存多个作业的整个管道

• Task Slot 是静态的概念，是指 TaskManager 具有的并发执行能力 。

下面看几个例子

并行可以分为两个方面

• 数据并行
  source 并行拉数据 map 并行处理数据

• 计算并行
  source 在拉新数据，map 在处理source 之前拉的数据
  两个 job 的并行执行

一个特定算子的 子任务（subtask）的个数被称之为其并行度（parallelism）。
一般情况下，一个 stream 的并行度，可以认为就是其所有算子中最大的并行度

idea里运行flink程序默认并行度是运行程序机器的核心数量。

每一个算子都可以单独设置并行。

.map((_, 1)).setParallelism(2)1

也可以全局指定并行度。

val env = ExecutionEnvironment.getExecutionEnvironment.setParallelism(2)此时不支持并行的算子 比如env.readTextFile(inputpath) 就会报错
具体情况调整source和sink的并行度123

三个位置可以配置并行度

• flink配置文件中

• 代码里

• flink任务提交时

优先级

代码>提交>配置文件

代码里设置用代码里的，代码里没设置用提交时设置的，都没设置用配置文件中的配置。
代码里算子单独设置优先级高于全局设置优先级

可以设置共享组 把 task 尽量均匀的分配到整个集群中

任务链
合理的设置并行度

• 减少本地通信的开销

• 减少序列化和反序列化

把多个算子合并为一个task，原本的算子成为里面的subtask

满足任务链需要一下条件

• 算子具有相同并行度(具有相同的分区数)

• 算子属于one-to-one

one-to-one ：stream维护着分区以及元素的顺序（比如source和map之间）。这意味着map 算子的子任务看到的元素的个数以及顺序跟 source 算子的子任务生产的元素的个数、顺序相同。map、fliter、flatMap等算子都是one-to-one的对应关系。

Redistributing：stream的分区会发生改变。每一个算子的子任务依据所选择的transformation发送数据到不同的目标任务。例如，keyBy 基于 hashCode 重分区、而 broadcast 和 rebalance 会随机重新分区，这些算子都会引起redistribute过程，而 redistribute 过程就类似于 Spark 中的 shuffle 过程。

并行度不同的算子之前传递数据会进行重分区，Redistributing类型的算子也会进行重分区。

• 例子
  
  配置文件中默认并行度设置为2 ，提交代码是并行度设置为2
  socket source 并行度只能是1
  flatmap fliter map 并行度都是2 且属于one-to-one 合成任务链
  keyby 属于redistrubuting hash 重分区
  sum print 并行度为2 属于one-to-one

执行图如下

当然还可以禁止掉合成任务链

单个算子不参与合成任务链

.flatMap(_.split(" ")).disableChaining()1

从单个算子开启一个新的任务链

.startNewChain()1

全局不合成任务链

env.disableOperatorChaining()1

下面是一个全局不合成任务链的job执行图,只是在上一个例子的基础上添加了全局不合成任务链。

算子设置并行度

• source 文件保证数顺序需要并行度为 1
  
  

• sink 只输出到一个文件需要并行度为 1
  

• socketsource 并行度只能为1

Operator Chains(操作链)

• Flink出于分布式执行的目的，将operator的subtask链接在一起形成task（类似spark中的管道）。

• 每个task在一个线程中执行。

• 将operators链接成task是非常有效的优化：它可以减少线程与线程间的切换和数据缓冲的开销，并在降低延迟的同时提高整体吞吐量。

• 链接的行为可以在编程API中进行指定,详情请见代码OperatorChainTest。

• 开启操作链 和 禁用操作链的对比图(默认开启):

  

  

• Flink默认会将多个operator进行串联，形成任务链(task chain)

• 注意: task chain 可以理解为就是 operator chain 只是不同场景下，称呼不同。

• 我们也可以禁用任务链，让每个operator形成一个task。

• StreamExecutionEnvironment.disableOperatorChaining() 这个方法会禁用整条工作链

• 操作链其实就是类似spark的pipeline管道模式，一个task可以执行同一个窄依赖中的算子操作。

• 我们也可以细粒度的控制工作链的形成，比如调用dataStreamSource.map(...).startNewChain(),但不能使用dataStreamSource.startNewChain()

• dataStreamSource.filter(...).map(...).startNewChain().map(...)，需要注意的是，当这样写时相当于source和filter组成一条链，两个map组成一条链。

• 即在filter和map之间断开，各自形成单独的链。

• 代码:

  package com.ronnie.flink.stream.test;import org.apache.flink.api.common.functions.MapFunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/**
   *  开启与禁用工作链时，输出的结果不一样。
   *  当开启工作链时(默认启动)，operator map1与map2 组成一个task.
   *     此时task运行时，对于hello，flink 这两条数据是：
   *     先打印 hello ---- 1 , hello->1 ---- 2
   *     后打印 flink ---- 1 , flink->1 ---- 2
   *  当禁用工作链时，operator map1与map2 分别在两个task中执行
   *     此时task运行时，对于hello，flink 这两条数据是：
   *     先打印 hello ---- 1 , flink ---- 1
   *     后打印 hello->1 ---- 2  , flink->1 ---- 2
   *
   *  注：操作链类似spark的管道,一个task执行多个的算子.
   */public class OperatorChainTest {    public static final String[] WORDS = new String[] {            "hello",            "flink",            "spark",            "hbase"
      };    public static void main(String[] args) {        // 设置执行环境, 类似spark中初始化sparkContext一样
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
          env.setParallelism(1);        // 关闭操作链..
          env.disableOperatorChaining();
  
          DataStreamSource dataStreamSource = env.fromElements(WORDS);
  
          SingleOutputStreamOperator pairStream = dataStreamSource.map(new MapFunction() {            @Override
              public String map(String value) throws Exception {
                  System.err.println(value + " ---- 1");                return value + "->1";
              }
          }).map(new MapFunction() {            @Override
              public String map(String value) throws Exception {
                  System.err.println(value + " ---- 2");                return value + "->2";
              }
          });        // 还可以控制更细粒度的任务链,比如指明从哪个operator开始形成一条新的链
          // someStream.map(...).startNewChain()，但不能使用someStream.startNewChain()。
          try {
              env.execute();
          } catch (Exception e) {
              e.printStackTrace();
          }
      }
  }

Task slots(任务槽)

• TaskManager 是一个 JVM 进程，并会以独立的线程来执行一个task或多个subtask。

• 为了控制一个 TaskManager 能接受多少个 task，Flink 提出了 Task Slot 的概念。

• Flink 中的计算资源通过 Task Slot 来定义。每个 task slot 代表了 TaskManager 的一个固定大小的资源子集。

• 例如，一个拥有3个slot的 TaskManager，会将其管理的内存平均分成三分分给各个 slot。

• 将资源 slot 化意味着来自不同job的task不会为了内存而竞争，而是每个task都拥有一定数量的内存储备。

• 需要注意的是，这里不会涉及到CPU的隔离，slot目前仅仅用来隔离task的内存。

• 通过调整 task slot 的数量，用户可以定义task之间是如何相互隔离的。

• 每个 TaskManager 有一个slot，也就意味着每个task运行在独立的 JVM 中。

• 每个 TaskManager 有多个slot的话，也就是说多个task运行在同一个JVM中。

• 而在同一个JVM进程中的task，可以共享TCP连接（基于多路复用）和心跳消息，可以减少数据的网络传输。

• 也能共享一些数据结构，一定程度上减少了每个task的消耗。

• 如图中所示，5个Task可能会在TaskManager的slots中分布，图中共2个TaskManager，每个有3个slot。