业务场景：
1.Flink 1.8.0 版本， Hadoop version: 3.0.0 ，
启动指令： /data/app/flink-1.8.0/bin/flink run -m yarn-cluster -p 4 ***.jar
先看异常的类型：
1.Flink App抛出The assigned slot container_e08_1539148828017_15937_01_003564_0 was removed.
2.Flink 提交任务抛出： java.lang.OutOfMemoryError
讨论：
1.首先考虑为什么会出现这两个异常，重点讨论第一个异常！
第一个异常，从异常的描述中我们很难看出问题的所在，一般情况下 container 是yarn资源调度平台的资源最小分配单位。而slot 是则是flink 程序中运行任务的一个通道，也就是说 正常情况下的 Flink on yarn 是这样运行的。 yarn 资源调度平台创建出 container ，然后 slot 就运行在 container 上，而flink 划分的一个个的任务就运行在 slot 上。如图所示
这里图上没有画出container ,但是我们要理解这个关系，要是还是不理解不了那再接下来看这个图：
所有 我们大致能推算出，既然任务在 container 中运行时被移除了， 那很大可能性就是 运行过程中所需的资源不够了，这时候有人会问，我的yarn集群资源很足啊，怎么可能资源不够了呢？ 这个问题其时也很好回答，你仔细想想，Flink 任务和 Yarn资源调度平台是什么关系？ 就是一个街坊邻居的关系，Flink 不过是借用 Yarn 的平台来运行自己的代码而已， Yarn 自己如何能判断出Flink 程序需要的资源量呢？ 其他Yarn是根据 Flink 提交任务指令和 flink-conf.yaml 中指定的默认配置来决定的。
既然明白这一点我们来看下这个指令： /data/app/flink-1.8.0/bin/flink run -m yarn-cluster -p 4 ***.jar 这个指定只是指定了 -p 的程序运行的并行度而已，并未指定所需的资源数量。所以Flink 程序运行的时候会根据 flink-conf.yaml 中的配置来划分资源的大小 并行度为4 就会运行 4个 slot.
看 flink-conf.yaml 中的内存指定，不难看出 一个TaskManager 指定4094m 共为32个slots ，不难算出 一个slot 所占的资源 128m,那这个资源到底够不够你的任务运行的呢？
这个需要你自己来衡量了，所有后来我通过指定：
/data/app/flink-1.8.0/bin/flink run -m yarn-cluster -p 4 -yn 2 ***.jar -ytm 6144
这时在运行的时候 就没有了这个异常了。
但是有的时候我将并行度指定为 1 ，会出现什么问题了 ， 程序很可能直接报异常 java.lang.OutOfMemoryError，因为什么单个slot的内存太小了 不足以启动这个程序，所以直接报内存溢出了，很明显是你指定的机器内存太小了 。这异常常常发生在测试的时候程序没有任务问题，一旦放到生产就会经常碰到。因为生产环境的数据量一般都要比测试大的多，所以会经常出现这种内存的问题。
