1.all to all的边的话，你这个例子把并发降下来肯定连接数要少很多的。 slot sharing的话也只会把A和B相同并发的share在一起，连接其他的subtask A还是要建立连接。 2.指的是作业jar,每个TM只会下载一次
Best regards, Weijie yidan zhao <hinobl...@gmail.com> 于2022年10月31日周一 19:54写道： > 嗯，问题1我主要是在想，这种复杂的连接关系，会不会增大Sending the partition request to '...' > failed；这种异常的概率。 > 问题2，你提到的下载jar是指任务jar还是flink的jar。flink的jar不需要，因为我是standalone集群。 > 任务jar的话，这出现另外一个问题，如果一个TM分配到120*10=1200个task，那么任务jar不会分发这么多次吧。 > > weijie guo <guoweijieres...@gmail.com> 于2022年10月31日周一 12:54写道： > > > > 你好，请问使用的flink版本是多少？ > > 1.15的话TM间是有connection reuse的，默认TM间建立一个物理TCP连接。 > > > 并发大了的话，你的TM只有一个slot，启动的TM会变多。task全变成running的状态变慢的因素也比较多：有些TM容器在的结点比较慢、下载jar包时间长、state > > restore慢等 > > > > Best regards, > > > > Weijie > > > > > > yidan zhao <hinobl...@gmail.com> 于2022年10月30日周日 11:36写道： > > > > > 如题，我生产集群频繁报 org.apache.flink.runtime.io > > > .network.netty.exception.LocalTransportException > > > 异常，具体异常cause有如下情况，按照出现频率从高到底列举。 > > > （1）Sending the partition request to '...' failed； > > > org.apache.flink.runtime.io > > > .network.netty.exception.LocalTransportException: > > > Sending the partition request to '/10.35.215.18:2065 (#0)' failed. > > > at org.apache.flink.runtime.io > > > > .network.netty.NettyPartitionRequestClient$1.operationComplete(NettyPartitionRequestClient.java:145) > > > at org.apache.flink.runtime.io > > > > .network.netty.NettyPartitionRequestClient$1.operationComplete(NettyPartitionRequestClient.java:133) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:578) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:552) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:491) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:616) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:609) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:117) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannel$AbstractUnsafe.safeSetFailure(AbstractChannel.java:1017) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannel$AbstractUnsafe.write(AbstractChannel.java:878) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline$HeadContext.write(DefaultChannelPipeline.java:1367) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:717) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeWriteAndFlush(AbstractChannelHandlerContext.java:764) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext$WriteTask.run(AbstractChannelHandlerContext.java:1071) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:469) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:384) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) > > > at java.lang.Thread.run(Thread.java:748) > > > Caused by: > > > > org.apache.flink.shaded.netty4.io.netty.channel.StacklessClosedChannelException > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannel$AbstractUnsafe.write(Object, > > > ChannelPromise)(Unknown Source) > > > Caused by: > > > > org.apache.flink.shaded.netty4.io.netty.channel.unix.Errors$NativeIoException: > > > writeAddress(..) failed: Connection timed out > > > > > > （2）readAddress(..) failed: Connection timed out > > > org.apache.flink.runtime.io > > > .network.netty.exception.LocalTransportException: > > > readAddress(..) failed: Connection timed out (connection to > > > '10.35.109.149/10.35.109.149:2094') > > > at org.apache.flink.runtime.io > > > > .network.netty.CreditBasedPartitionRequestClientHandler.exceptionCaught(CreditBasedPartitionRequestClientHandler.java:168) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:302) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:281) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireExceptionCaught(AbstractChannelHandlerContext.java:273) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline$HeadContext.exceptionCaught(DefaultChannelPipeline.java:1377) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:302) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:281) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline.fireExceptionCaught(DefaultChannelPipeline.java:907) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.handleReadException(AbstractEpollStreamChannel.java:728) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:821) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:480) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:378) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986) > > > at > > > > org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) > > > at java.lang.Thread.run(Thread.java:748) > > > Caused by: > > > > org.apache.flink.shaded.netty4.io.netty.channel.unix.Errors$NativeIoException: > > > readAddress(..) failed: Connection timed out > > > > > > 集群信息：standalone 集群，上百台机器。每台都是自研容器，10g内存左右，每个TM提供1个slot，避免任务之间影响。 > > > > > > > 作业信息：作业数小于10个。复杂的单任务大概用120并发执行，流量入口16w/s左右，数据处理、5min平滑window处理（目前偶尔buzz在window处）、redis > > > 异步算子等。 > > > > > > > 作业状态：反压目前倒是不严重，基本很少反压了。但经常因为出现如上的连接相关问题，自研容器虽然存在故障重启等可能性，但经过调整当前容器本身出问题概率多天才1次，可是任务可能一天好多次报LocalTransportException异常。 > > > > > > > 此外，按照之前的经验，LocalTransportException的出现概率和我的集群运行时间也有关，长期运行大概几天后就会加大概率，然后整体重启后会好转。 > > > > > > > > > > 当前在考虑优化可能方向，大佬们看看有啥建议，实际因为做实验对比周期太长，最好有一定倾向，否则控制影响因素一个一个对比，而且每次对比需要长期观察，时间太久。 > > > （1）slot group分到不同slot group，然后降低整体并发。 > > > 如果问题是tm的task之间连接对太多导致，这个可能会有效。比如A=>B的120并发，改成60并发的A=>B，让AB处于不同slot > > > group。 > > > 从连接对角度好像降低了复杂度，但单容器流量可能会增大（因为大流量的算子可能集中某60个容器，另外60个容器流量变小可能）。 > > > 其次，如果将任务分为3个slot > > > group，按60并发，则实际需180个容器。180*每个容器发现故障的概率，一定程度上增大了因容器故障导致的作业重启问题。 > > > 注意如上的连接对我理解其实也是“逻辑概念”，因为tm之间tcp连接是共享的？所以实际‘连接对’也只是 > > > resultSubPartition 的数量问题。 > > > > > > > （2）taskmanager.network.netty.num-arenas、taskmanager.network.netty.server.numThreads、taskmanager.network.netty.client.numThreads > > > 这三参数是否有影响呢，由于tm的slot为1，这3参数默认都是slot数，也就是1。 > > > > > > > > > > > > > 此外，还有一个问题，并发大了是否会影响任务启动过程速度。我今天启动120并发的任务，task从deploying变成initializing再到running是一点点变的，差不多1-2min才全部变成running。 > > > >
