title: Spark 2.2 Sort fails with huge dataset
body: <p>I am facing an issue when <strong>sorting</strong> a huge dataset (<strong>1.2 T</strong>) based on 4 columns. I also need right after the sort, to partition this dataset when writing the final dataset in HDFS ,based on one of the columns used in the sort function.</p>
  <p>Here is a stackoverflow post I posted a few days ago describing an other issue I had with the same code but with regards to joining two datasets :</p>
  <p><a href="https://stackoverflow.com/questions/55919699/spark-2-2-join-fails-with-huge-dataset">previous issue</a></p>
  <p>I used the answer of this post to improve my code. Now the join works fine.</p>
  <p>I tested the code without the sort and it works fine. In order to perform the sort, I thought about partitioning the data based on the four columns.</p>
  <p>The size of one partition is <strong>500MB</strong>. I have then <code>2600=1.2T/500MB</code> partitions.</p>
  <p>When executing the spark job, I get an <code>shuffle.RetryingBlockFetcher</code> error (<em>see the error logs below</em>).</p>
  <p>My <strong>questions</strong> are :</p>
  <ul>
  <li>What is the best way to sort data in spark to avoid shuffles ? Or reducing it ?</li>
  <li>Could I correct/add improvements to my code in order to perform the sort ?</li>
  <li>Do I really have to sort this way ? Can't I use other techniques like a Group By ?</li>
  </ul>
  <h3>Snippet of my code :</h3>
  <h1>EDITED</h1>
  <pre><code>    val uh = uh_months
        .withColumn(&quot;UHDIN&quot;, datediff(to_date(unix_timestamp(col(&quot;UHDIN_YYYYMMDD&quot;), &quot;yyyyMMdd&quot;).cast(TimestampType)),
          to_date(unix_timestamp(col(&quot;january&quot;), &quot;yyyy-MM-dd&quot;).cast(TimestampType))))
        //      .withColumn(&quot;DVA_1&quot;, to_date((unix_timestamp(col(&quot;DVA&quot;), &quot;ddMMMyyyy&quot;)).cast(TimestampType)))
        .withColumn(&quot;DVA_1&quot;, date_format(col(&quot;DVA&quot;), &quot;dd/MM/yyyy&quot;))
        .drop(&quot;UHDIN_YYYYMMDD&quot;)
        .drop(&quot;january&quot;)
        .drop(&quot;DVA&quot;).repartition(1300,col(&quot;MMED&quot;),col(&quot;DEBCRED&quot;),col(&quot;NMTGP&quot;))//.repartition(1300,col(&quot;NO_NUM&quot;))
  
      val uh_flag_comment = new TransactionType().transform(uh)
      val uh_repartitioned = uh_flag_comment.repartition(1300,col(&quot;NO_NUM&quot;))
  
      val uh_joined = uh_repartitioned.join(broadcast(smallDF), &quot;NO_NUM&quot;)
        .select(
          uh.col(&quot;*&quot;),
          smallDF.col(&quot;PSP&quot;),
          smallDF.col(&quot;minrel&quot;),
          smallDF.col(&quot;Label&quot;),
          smallDF.col(&quot;StartDate&quot;))
        .withColumnRenamed(&quot;DVA_1&quot;, &quot;DVA&quot;)
  
      val uh_final = uh_joined.repartition(1300, col(&quot;PSP&quot;)).sortWithinPartitions(col(&quot;NO_NUM&quot;), col(&quot;UHDIN&quot;), col(&quot;HOURMV&quot;))
  
      return uh_final
  </code></pre>
  <p><code>TransactionType</code> is a class where I add a new column to my <code>uh</code> dataframe based on the value of 3 columns (<code>MMED</code>, <code>DEBCRED</code>, <code>NMTGP</code>), using regex.</p>
  <p>Without the sort, and using the full capacity of the cluster, the code runs within 1 hour approximately.</p>
  <h3>Execution plan</h3>
  <pre><code>== Physical Plan ==
  Exchange hashpartitioning(PSP#82, 2600)
  +- *Sort [PSP#82 ASC NULLS FIRST, NO_NUM#252 ASC NULLS FIRST, UHDIN#547 ASC NULLS FIRST, HOURMV#175 ASC NULLS FIRST], true, 0
     +- Exchange rangepartitioning(PSP#82 ASC NULLS FIRST, NO_NUM#252 ASC NULLS FIRST, UHDIN#547 ASC NULLS FIRST, HOURMV#175 ASC NULLS FIRST, 200)
        +- Exchange hashpartitioning(PSP#82, NO_NUM#252, UHDIN#547, HOURMV#175, 2600)
           +- *Project [NO_NUM#252, DEV#153, DEBCRED#154, BDGRORI#155, BDGREUR#156, BEWC#157, MSG30_NL#158, SCAPMV#159, USERID#160, MMED#161, TNUM#162, NMTGP#163, BKA#164, CATEXT#165, SEQETAT#166, ACCTYPE#167, BRAND#168, FAMILY#169, SUBFAMILY#170, FORCED_DVA#172, BYBANK#173, CPTE_PROTEGE#174, HOURMV#175, RDFB#176, ... 30 more fields]
              +- *BroadcastHashJoin [NO_NUM#252], [NO_NUM#13], Inner, BuildRight
                 :- Exchange hashpartitioning(NO_NUM#252, 1300)
                 :  +- *Project [NUM#152 AS NO_NUM#252, DEV#153, DEBCRED#154, BDGRORI#155, BDGREUR#156, BEWC#157, MSG30_NL#158, SCAPMV#159, USERID#160, MMED#161, TNUM#162, NMTGP#163, BKA#164, CATEXT#165, SEQETAT#166, ACCTYPE#167, BRAND#168, FAMILY#169, SUBFAMILY#170, FORCED_DVA#172, BYBANK#173, CPTE_PROTEGE#174, HOURMV#175, RDFB#176, ... 26 more fields]
                 :     +- *Filter (BEWC#157 INSET (25003,25302,25114,20113,12017,20108,25046,12018,15379,15358,11011,20114,10118,12003,25097,20106,20133,10133,10142,15402,25026,25345,28023,15376,25019,28004,21701,25001,11008,15310,15003,2SOMEPORT,22048,15470,25300,25514,25381,25339,15099,25301,28005,28026,25098,25018,15323,25376,15804,15414,25344,25102,15458,15313,28002,25385,22051,25214,15031,12005,15425,20145,22011,15304,25027,14020,11007,25901,15343,22049,20112,12031,20127,15339,25421,15432,28025,25340,25325,20150,28011,25368,25304,22501,25369,28022,15098,12032,15375,25002,25008,10116,10101,22502,25090,15004,20105,12030,22503,15095,22007,15809,15342,15311,25216,10103,20122,11019,20142,15097,20147,20149,25005,25205,25380,15380,10120,25015,15384,11003,10110,25016,15090,25307,15001,25390,15312,10115,25219,15806,15459,12016,15359,15395,15302,12021,11701,10111,10148,25379,15807,10102,25352,25355,12010,25095,25394,20101,25413,15385,25322,28027,11026,15533,25201,25371,10128,11028,12020,15819,10143,28028,10123,10125,11020,25029,10122,25343,15015,12033,25014,12012,25024,25375,11023,25501,25402,22001,15317,12014,16114,20501,15046,12001,12022,10104,10117,12002,25499,10145,10153,12011,15350,15300,10119,25305,15345,25374,11027,25430,28021,25202,10121,28024,25101,28001,15321,11025,25358,15333,15501,25533,15372,12008,11015,10114,10113,10112,15303,15320,28006,22002,25359,10132,15497,25353,11029,25425,15374,12019,25437,11022,15357,20148,20111,26114,25099,25354,10124,25303,11010,20120,20135,15820,15331,28029) &amp;&amp; isnotnull(NUM#152))
                 :        +- *FileScan csv [UHDIN_YYYYMMDD#151,NUM#152,DEV#153,DEBCRED#154,BDGRORI#155,BDGREUR#156,BEWC#157,MSG30_NL#158,SCAPMV#159,USERID#160,MMED#161,TNUM#162,NMTGP#163,BKA#164,CATEXT#165,SEQETAT#166,ACCTYPE#167,BRAND#168,FAMILY#169,SUBFAMILY#170,DVA#171,FORCED_DVA#172,BYBANK#173,CPTE_PROTEGE#174,... 26 more fields] Batched: false, Format: CSV, Location: InMemoryFileIndex[SOMEHOST:SOMEPORT/SOMEPATH, PartitionFilters: [], PushedFilters: [In(BEWC, [25003,25302,25114,20113,12017,20108,25046,12018,15379,15358,11011,20114,10118,12003,25..., ReadSchema: struct&lt;UHDIN_YYYYMMDD:string,NUM:string,DEV:string,DEBCRED:string,BDGRORI:string,BDGREUR:string,B...
                 +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]))
                    +- *Project [NO_NUM#13, PSP#82, minrel#370, Label#105, StartDate#106]
                       +- *SortMergeJoin [PSP#381], [PSP#82], Inner
                          :- *Sort [PSP#381 ASC NULLS FIRST], false, 0
                          :  +- Exchange hashpartitioning(PSP#381, 200)
                          :     +- *Project [PSP#381, NO_NUM#13, minrel#370]
                          :        +- SortMergeJoin [PSP#381, C_SNUM#14, minrel#370, NO_NUM#13], [NO_PSP#47, C_SNUM_1#387, C_NRELPR#50, NO_NUM_1#400], LeftOuter
                          :           :- *Sort [PSP#381 ASC NULLS FIRST, C_SNUM#14 ASC NULLS FIRST, minrel#370 ASC NULLS FIRST, NO_NUM#13 ASC NULLS FIRST], false, 0
                          :           :  +- Exchange hashpartitioning(PSP#381, C_SNUM#14, minrel#370, NO_NUM#13, 200)
                          :           :     +- SortAggregate(key=[NO_PSP#12, C_SNUM#14, NO_NUM#13], functions=[min(C_NRELPR#15)])
                          :           :        +- *Sort [NO_PSP#12 ASC NULLS FIRST, C_SNUM#14 ASC NULLS FIRST, NO_NUM#13 ASC NULLS FIRST], false, 0
                          :           :           +- Exchange hashpartitioning(NO_PSP#12, C_SNUM#14, NO_NUM#13, 200)
                          :           :              +- SortAggregate(key=[NO_PSP#12, C_SNUM#14, NO_NUM#13], functions=[partial_min(C_NRELPR#15)])
                          :           :                 +- *Sort [NO_PSP#12 ASC NULLS FIRST, C_SNUM#14 ASC NULLS FIRST, NO_NUM#13 ASC NULLS FIRST], false, 0
                          :           :                    +- *Project [NO_PSP#12, C_SNUM#14, NO_NUM#13, C_NRELPR#15]
                          :           :                       +- *Filter (((C_NRELPR#15 IN (001,006) &amp;&amp; C_SNUM#14 IN (030,033)) &amp;&amp; isnotnull(NO_NUM#13)) &amp;&amp; isnotnull(NO_PSP#12))
                          :           :                          +- *FileScan csv [NO_PSP#12,NO_NUM#13,C_SNUM#14,c_nrelpr#15] Batched: false, Format: CSV, Location: InMemoryFileIndex[SOMEHOST:SOMEPORT/SOMEPATH, PartitionFilters: [], PushedFilters: [In(c_nrelpr, [001,006]), In(C_SNUM, [030,033]), IsNotNull(NO_NUM), IsNotNull(NO_PSP)], ReadSchema: struct&lt;NO_PSP:string,NO_NUM:string,C_SNUM:string,c_nrelpr:string&gt;
                          :           +- *Sort [NO_PSP#47 ASC NULLS FIRST, C_SNUM_1#387 ASC NULLS FIRST, C_NRELPR#50 ASC NULLS FIRST, NO_NUM_1#400 ASC NULLS FIRST], false, 0
                          :              +- Exchange hashpartitioning(NO_PSP#47, C_SNUM_1#387, C_NRELPR#50, NO_NUM_1#400, 200)
                          :                 +- *Project [NO_PSP#47, NO_NUM#48 AS NO_NUM_1#400, C_SNUM#49 AS C_SNUM_1#387, c_nrelpr#50]
                          :                    +- *FileScan csv [NO_PSP#47,NO_NUM#48,C_SNUM#49,c_nrelpr#50] Batched: false, Format: CSV, Location: InMemoryFileIndex[SOMEHOST:SOMEPORT/SOMEPATH, PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;NO_PSP:string,NO_NUM:string,C_SNUM:string,c_nrelpr:string&gt;
                          +- *Sort [PSP#82 ASC NULLS FIRST], false, 0
                             +- Exchange hashpartitioning(PSP#82, 200)
                                +- *Project [PSP#82, Label#105, StartDate#106]
                                   +- *Filter isnotnull(PSP#82)
                                      +- *FileScan csv [PSP#82,Label#105,StartDate#106] Batched: false, Format: CSV, Location: InMemoryFileIndex[SOMEHOST:SOMEPORT/SOMEPATH, PartitionFilters: [], PushedFilters: [IsNotNull(PSP)], ReadSchema: struct&lt;PSP:string,Label:string,StartDate:string&gt;
  
  </code></pre>
  <p>Here is the main error(s) I got when launching the job with the sort :</p>
  <pre><code>19/05/06 18:02:25 ERROR shuffle.RetryingBlockFetcher: Exception while beginning fetch of 214 outstanding blocks 
  java.io.IOException: Failed to connect to SOMEHOST/SOMEADDRESS:SOMEPORT
      at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:232)
      at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:182)
      at org.apache.spark.network.netty.NettyBlockTransferService$$anon$1.createAndStart(NettyBlockTransferService.scala:98)
      at org.apache.spark.network.shuffle.RetryingBlockFetcher.fetchAllOutstanding(RetryingBlockFetcher.java:141)
      at org.apache.spark.network.shuffle.RetryingBlockFetcher.start(RetryingBlockFetcher.java:121)
      at org.apache.spark.network.netty.NettyBlockTransferService.fetchBlocks(NettyBlockTransferService.scala:108)
      at org.apache.spark.storage.ShuffleBlockFetcherIterator.sendRequest(ShuffleBlockFetcherIterator.scala:228)
      at org.apache.spark.storage.ShuffleBlockFetcherIterator.fetchUpToMaxBytes(ShuffleBlockFetcherIterator.scala:435)
      at org.apache.spark.storage.ShuffleBlockFetcherIterator.initialize(ShuffleBlockFetcherIterator.scala:323)
      at org.apache.spark.storage.ShuffleBlockFetcherIterator.&lt;init&gt;(ShuffleBlockFetcherIterator.scala:140)
      at org.apache.spark.shuffle.BlockStoreShuffleReader.read(BlockStoreShuffleReader.scala:45)
      at org.apache.spark.sql.execution.ShuffledRowRDD.compute(ShuffledRowRDD.scala:165)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
      at org.apache.spark.scheduler.Task.run(Task.scala:108)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: SOMEHOST/SOMEADDRESS:SOMEPORT
      at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
      at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
      at io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:257)
      at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:291)
      at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:631)
      at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:566)
      at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:480)
      at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:442)
      at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:131)
      at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
      ... 1 more
  19/05/06 18:02:25 INFO shuffle.RetryingBlockFetcher: Retrying fetch (1/3) for 214 outstanding blocks after 5000 ms
  19/05/06 18:02:25 INFO storage.ShuffleBlockFetcherIterator: Started 6 remote fetches in 13 ms
  19/05/06 18:02:28 INFO executor.Executor: Finished task 408.0 in stage 14.0 (TID 6696). 1733 bytes result sent to driver
  19/05/06 18:02:28 INFO executor.CoarseGrainedExecutorBackend: Got assigned task 6816
  19/05/06 18:02:28 INFO executor.Executor: Running task 466.1 in stage 14.0 (TID 6816)
  19/05/06 18:02:28 INFO storage.ShuffleBlockFetcherIterator: Getting 5073 non-empty blocks out of 5089 blocks
  19/05/06 18:02:28 INFO client.TransportClientFactory: Found inactive connection to SOMEHOST/SOMEADDRESS:SOMEPORT, creating a new one.
  19/05/06 18:02:28 ERROR shuffle.RetryingBlockFetcher: Exception while beginning fetch of 82 outstanding blocks 
  java.io.IOException: Failed to connect to SOMEHOST/SOMEADDRESS:SOMEPORT
      at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:232)
      at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:182)
      at org.apache.spark.network.netty.NettyBlockTransferService$$anon$1.createAndStart(NettyBlockTransferService.scala:98)
      at org.apache.spark.network.shuffle.RetryingBlockFetcher.fetchAllOutstanding(RetryingBlockFetcher.java:141)
      at org.apache.spark.network.shuffle.RetryingBlockFetcher.start(RetryingBlockFetcher.java:121)
      at org.apache.spark.network.netty.NettyBlockTransferService.fetchBlocks(NettyBlockTransferService.scala:108)
      at org.apache.spark.storage.ShuffleBlockFetcherIterator.sendRequest(ShuffleBlockFetcherIterator.scala:228)
      at org.apache.spark.storage.ShuffleBlockFetcherIterator.fetchUpToMaxBytes(ShuffleBlockFetcherIterator.scala:435)
      at org.apache.spark.storage.ShuffleBlockFetcherIterator.initialize(ShuffleBlockFetcherIterator.scala:323)
      at org.apache.spark.storage.ShuffleBlockFetcherIterator.&lt;init&gt;(ShuffleBlockFetcherIterator.scala:140)
      at org.apache.spark.shuffle.BlockStoreShuffleReader.read(BlockStoreShuffleReader.scala:45)
      at org.apache.spark.sql.execution.ShuffledRowRDD.compute(ShuffledRowRDD.scala:165)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
      at org.apache.spark.scheduler.Task.run(Task.scala:108)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: SOMEHOST/SOMEADDRESS:SOMEPORT
      at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
      at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
      at io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:257)
      at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:291)
      at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:631)
      at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:566)
      at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:480)
      at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:442)
      at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:131)
      at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
      ... 1 more
  </code></pre>
  <h1>EDITED</h1>
  <p>An other type of error :</p>
  <pre><code>19/05/06 18:06:16 ERROR executor.Executor: Exception in task 309.1 in stage 13.1 (TID 7592)
  java.io.FileNotFoundException: /applis/hadoop/yarn/local/usercache/MYUSER/appcache/application_1555263602441_0123/blockmgr-aa586b76-ff58-4f88-b168-288c3e1b9f61/3c/temp_shuffle_ea967624-f633-4481-9a05-249b561e3c38 (No such file or directory)
      at java.io.FileInputStream.open0(Native Method)
      at java.io.FileInputStream.open(FileInputStream.java:195)
      at java.io.FileInputStream.&lt;init&gt;(FileInputStream.java:138)
      at org.spark_project.guava.io.Files$FileByteSource.openStream(Files.java:124)
      at org.spark_project.guava.io.Files$FileByteSource.openStream(Files.java:114)
      at org.spark_project.guava.io.ByteSource.copyTo(ByteSource.java:202)
      at org.spark_project.guava.io.Files.copy(Files.java:436)
      at org.spark_project.guava.io.Files.move(Files.java:651)
      at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.mergeSpills(UnsafeShuffleWriter.java:277)
      at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.closeAndWriteOutput(UnsafeShuffleWriter.java:216)
      at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:169)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
      at org.apache.spark.scheduler.Task.run(Task.scala:108)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)
  19/05/06 18:06:16 ERROR executor.Executor: Exception in task 502.1 in stage 13.1 (TID 7599)
  java.io.FileNotFoundException: /applis/hadoop/yarn/local/usercache/MYUSER/appcache/application_1555263602441_0123/blockmgr-aa586b76-ff58-4f88-b168-288c3e1b9f61/34/temp_shuffle_dd202cd1-ad8f-41c4-b4d1-d79621cd169e (No such file or directory)
      at java.io.FileOutputStream.open0(Native Method)
      at java.io.FileOutputStream.open(FileOutputStream.java:270)
      at java.io.FileOutputStream.&lt;init&gt;(FileOutputStream.java:213)
      at org.apache.spark.storage.DiskBlockObjectWriter.initialize(DiskBlockObjectWriter.scala:102)
      at org.apache.spark.storage.DiskBlockObjectWriter.open(DiskBlockObjectWriter.scala:115)
      at org.apache.spark.storage.DiskBlockObjectWriter.write(DiskBlockObjectWriter.scala:247)
      at org.apache.spark.shuffle.sort.ShuffleExternalSorter.writeSortedFile(ShuffleExternalSorter.java:201)
      at org.apache.spark.shuffle.sort.ShuffleExternalSorter.closeAndGetSpills(ShuffleExternalSorter.java:405)
      at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.closeAndWriteOutput(UnsafeShuffleWriter.java:209)
      at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:169)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
      at org.apache.spark.scheduler.Task.run(Task.scala:108)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)
  19/05/06 18:06:16 INFO executor.Executor: Finished task 200.2 in stage 13.1 (TID 7568). 2826 bytes result sent to driver
  19/05/06 18:06:16 ERROR executor.CoarseGrainedExecutorBackend: RECEIVED SIGNAL TERM
  19/05/06 18:06:16 INFO util.ShutdownHookManager: Shutdown hook called
  </code></pre>
  <h2>Some information/context:</h2>
  <p>I am working on a production environment (see the cluster configuration below). I cannot upgrade my spark version. I do not have spark UI or yarn UI to monitor my jobs. All I can retrieve are the yarn logs.</p>
  <h3>Spark Version: 2.2</h3>
  <h3>Cluster configuration:</h3>
  <ul>
  <li>21 compute nodes (workers)</li>
  <li>8 cores each</li>
  <li>64 GB RAM per node</li>
  </ul>
  <h3>Current Spark configuration:</h3>
  <p>-master: yarn</p>
  <p>-executor-memory: 42G</p>
  <p>-executor-cores: 5</p>
  <p>-driver memory: 42G</p>
  <p>-num-executors: 32</p>
  <p>-spark.sql.broadcastTimeout=3600</p>
  <p>-spark.kryoserializer.buffer.max=512</p>
  <p>-spark.yarn.executor.memoryOverhead=2400</p>
  <p>-spark.driver.maxResultSize=500m</p>
  <p>-spark.memory.storageFraction=0.3</p>
  <p>-spark.memory.fraction=0.9</p>
  <p>-spark.hadoop.fs.permissions.umask-mode=007</p>
  <h3>How is the job executed:</h3>
  <p>We build an artifact (jar) with IntelliJ and then send it to a server. Then a bash script is executed. This script:</p>
  <ul>
  <li><p>export some environment variables (SPARK_HOME, HADOOP_CONF_DIR, PATH and SPARK_LOCAL_DIRS)</p>
  </li>
  <li><p>launch the spark-submit command with all the parameters defined in the spark configuration above</p>
  </li>
  <li><p>retrieves the yarn logs of the application</p>
  </li>
  </ul>