title: Databricks Notebook 8.3 (Apache Spark 3.1.1, Scala 2.12) | pyspark | Parquet write exception | Multiple failures in stage materialization
body: <p>This is a Production code running fine until last week. Then, this parquet write error showed up and never getting resolved.</p>
  <p>While writing to AWS S3 in parquet format, I tried several dataframe.repartitions(300) - 300, 500, 2400, 6000. But no luck. The code by itself runs fine, but some times gives count error if I add a count() on a dataframe. (intermittently).
  So I removed all the count()s in the code to run the code without errors. Now, It fails while writing to the AWS s3 location.</p>
  <p>The code is running on Databricks notebook - Databricks Runtime Version
  8.3 (includes Apache Spark 3.1.1, Scala 2.12). The code is written in pyspark(python 3.8). The code runs on AWS r5.8xlarge instances.
  I am stuck with this, any help is very much appreciated.</p>
  <pre><code>Py4JJavaError                             Traceback (most recent call last)
  &lt;command-2026517708936858&gt; in &lt;module&gt;
        3 
        4 #save data_agg for next step
  ----&gt; 5 dataframe.repartition(6000).write.parquet(s3://path_to_write, mode='overwrite')
  
  /databricks/spark/python/pyspark/sql/readwriter.py in parquet(self, path, mode, partitionBy, compression)
     1275             self.partitionBy(partitionBy)
     1276         self._set_opts(compression=compression)
  -&gt; 1277         self._jwrite.parquet(path)
     1278 
     1279     def text(self, path, compression=None, lineSep=None):
  
  /databricks/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py in __call__(self, *args)
     1302 
     1303         answer = self.gateway_client.send_command(command)
  -&gt; 1304         return_value = get_return_value(
     1305             answer, self.gateway_client, self.target_id, self.name)
     1306 
  
  /databricks/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
      115     def deco(*a, **kw):
      116         try:
  --&gt; 117             return f(*a, **kw)
      118         except py4j.protocol.Py4JJavaError as e:
      119             converted = convert_exception(e.java_exception)
  
  /databricks/spark/python/lib/py4j-0.10.9-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
      324             value = OUTPUT_CONVERTER[type](answer[2:], gateway_client)
      325             if answer[1] == REFERENCE_TYPE:
  --&gt; 326                 raise Py4JJavaError(
      327                     &quot;An error occurred while calling {0}{1}{2}.\n&quot;.
      328                     format(target_id, &quot;.&quot;, name), value)
  
  Py4JJavaError: An error occurred while calling o1773.parquet.
  : org.apache.spark.SparkException: Job aborted.
      at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:289)
      at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand.run(InsertIntoHadoopFsRelationCommand.scala:203)
      at org.apache.spark.sql.execution.command.DataWritingCommandExec.sideEffectResult$lzycompute(commands.scala:121)
      at org.apache.spark.sql.execution.command.DataWritingCommandExec.sideEffectResult(commands.scala:119)
      at org.apache.spark.sql.execution.command.DataWritingCommandExec.doExecute(commands.scala:144)
      at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:196)
      at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:240)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:165)
      at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:236)
      at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:192)
      at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:167)
      at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:166)
      at org.apache.spark.sql.DataFrameWriter.$anonfun$runCommand$1(DataFrameWriter.scala:1079)
      at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withCustomExecutionEnv$5(SQLExecution.scala:126)
      at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:267)
      at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withCustomExecutionEnv$1(SQLExecution.scala:104)
      at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:852)
      at org.apache.spark.sql.execution.SQLExecution$.withCustomExecutionEnv(SQLExecution.scala:77)
      at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:217)
      at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:1079)
      at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:468)
      at org.apache.spark.sql.DataFrameWriter.saveInternal(DataFrameWriter.scala:438)
      at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:303)
      at org.apache.spark.sql.DataFrameWriter.parquet(DataFrameWriter.scala:964)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:498)
      at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
      at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:380)
      at py4j.Gateway.invoke(Gateway.java:295)
      at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
      at py4j.commands.CallCommand.execute(CallCommand.java:79)
      at py4j.GatewayConnection.run(GatewayConnection.java:251)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: org.apache.spark.SparkException: Multiple failures in stage materialization.
      at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.cleanUpAndThrowException(AdaptiveSparkPlanExec.scala:838)
      at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.$anonfun$getFinalPhysicalPlan$1(AdaptiveSparkPlanExec.scala:321)
      at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:852)
      at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.getFinalPhysicalPlan(AdaptiveSparkPlanExec.scala:276)
      at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.doExecute(AdaptiveSparkPlanExec.scala:378)
      at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:196)
      at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:240)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:165)
      at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:236)
      at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:192)
      at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:233)
      ... 34 more
      Suppressed: org.apache.spark.SparkException: Job aborted due to stage failure: ShuffleMapStage 46 (parquet at NativeMethodAccessorImpl.java:0) has failed the maximum allowable number of times: 4. Most recent failure reason: org.apache.spark.shuffle.FetchFailedException: Connecting to /100.64.19.5:4048 failed in the last 4750 ms, fail this connection directly
      at org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:771)
      at org.apache.spark.storage.ShuffleBlockFetcherIterator.$anonfun$next$1(ShuffleBlockFetcherIterator.scala:686)
      at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
      at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:577)
      at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:70)
      at org.apache.spark.util.CompletionIterator.next(CompletionIterator.scala:29)   at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:484)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:490)
      at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
      at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31)
      at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
      at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage14.agg_doAggregateWithKeys_0$(Unknown Source)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage14.sort_addToSorter_0$(Unknown Source)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage14.processNext(Unknown Source)
      at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
      at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:757)
      at org.apache.spark.sql.execution.RowIteratorFromScala.advanceNext(RowIterator.scala:83)
      at org.apache.spark.sql.execution.joins.SortMergeFullOuterJoinScanner.advancedLeft(SortMergeJoinExec.scala:1088)
      at org.apache.spark.sql.execution.joins.SortMergeFullOuterJoinScanner.&lt;init&gt;(SortMergeJoinExec.scala:1078)
      at org.apache.spark.sql.execution.joins.SortMergeJoinExec.$anonfun$doExecute$1(SortMergeJoinExec.scala:222)
      at org.apache.spark.rdd.ZippedPartitionsRDD2.compute(ZippedPartitionsRDD.scala:125)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)  at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
      at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
      at org.apache.spark.scheduler.ShuffleMapTask.$anonfun$runTask$3(ShuffleMapTask.scala:81)
      at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
      at org.apache.spark.scheduler.ShuffleMapTask.$anonfun$runTask$1(ShuffleMapTask.scala:81)
      at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
      at org.apache.spark.scheduler.Task.doRunTask(Task.scala:150)
      at org.apache.spark.scheduler.Task.$anonfun$run$1(Task.scala:119)
      at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
      at org.apache.spark.scheduler.Task.run(Task.scala:91)
      at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$13(Executor.scala:812)
      at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1643)
      at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:815)
      at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
      at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:671)
      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: java.io.IOException: Connecting to /100.64.19.5:4048 failed in the last 4750 ms, fail this connection directly
      at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:214)
      at org.apache.spark.network.shuffle.ExternalBlockStoreClient.lambda$fetchBlocks$0(ExternalBlockStoreClient.java:101)
      at org.apache.spark.network.shuffle.RetryingBlockFetcher.fetchAllOutstanding(RetryingBlockFetcher.java:153)
      at org.apache.spark.network.shuffle.RetryingBlockFetcher.lambda$initiateRetry$0(RetryingBlockFetcher.java:181)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)  at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
      ... 1 more
   
          at org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2765)
          at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2712)
          at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2706)
          at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
          at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
          at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
          at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2706)
          at org.apache.spark.scheduler.DAGScheduler.handleTaskCompletion(DAGScheduler.scala:2263)
          at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2970)
          at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2914)
          at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2902)
          at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
      Suppressed: [CIRCULAR REFERENCE: org.apache.spark.SparkException: Job aborted due to stage failure: ShuffleMapStage 46 (parquet at NativeMethodAccessorImpl.java:0) has failed the maximum allowable number of times: 4. Most recent failure reason: org.apache.spark.shuffle.FetchFailedException: Connecting to /100.64.19.5:4048 failed in the last 4750 ms, fail this connection directly   at org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:771)    at org.apache.spark.storage.ShuffleBlockFetcherIterator.$anonfun$next$1(ShuffleBlockFetcherIterator.scala:686)  at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)     at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:577)     at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:70)  at org.apache.spark.util.CompletionIterator.next(CompletionIterator.scala:29)   at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:484)   at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:490)   at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)   at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31)    at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)   at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)   at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage14.agg_doAggregateWithKeys_0$(Unknown Source)   at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage14.sort_addToSorter_0$(Unknown Source)  at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage14.processNext(Unknown Source)  at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)  at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:757)    at org.apache.spark.sql.execution.RowIteratorFromScala.advanceNext(RowIterator.scala:83)    at org.apache.spark.sql.execution.joins.SortMergeFullOuterJoinScanner.advancedLeft(SortMergeJoinExec.scala:1088)    at org.apache.spark.sql.execution.joins.SortMergeFullOuterJoinScanner.&lt;init&gt;(SortMergeJoinExec.scala:1078)  at org.apache.spark.sql.execution.joins.SortMergeJoinExec.$anonfun$doExecute$1(SortMergeJoinExec.scala:222)     at org.apache.spark.rdd.ZippedPartitionsRDD2.compute(ZippedPartitionsRDD.scala:125)     at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)  at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)     at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)     at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)  at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)     at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)     at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)  at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)     at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)     at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)  at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)     at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)     at org.apache.spark.scheduler.ShuffleMapTask.$anonfun$runTask$3(ShuffleMapTask.scala:81)    at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)     at org.apache.spark.scheduler.ShuffleMapTask.$anonfun$runTask$1(ShuffleMapTask.scala:81)    at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)     at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)   at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)   at org.apache.spark.scheduler.Task.doRunTask(Task.scala:150)    at org.apache.spark.scheduler.Task.$anonfun$run$1(Task.scala:119)   at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)     at org.apache.spark.scheduler.Task.run(Task.scala:91)   at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$13(Executor.scala:812)    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1643)    at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:815)     at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)   at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)     at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:671)    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: java.io.IOException: Connecting to /100.64.19.5:4048 failed in the last 4750 ms, fail this connection directly  at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:214)     at org.apache.spark.network.shuffle.ExternalBlockStoreClient.lambda$fetchBlocks$0(ExternalBlockStoreClient.java:101)    at org.apache.spark.network.shuffle.RetryingBlockFetcher.fetchAllOutstanding(RetryingBlockFetcher.java:153)     at org.apache.spark.network.shuffle.RetryingBlockFetcher.lambda$initiateRetry$0(RetryingBlockFetcher.java:181)  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)  at java.util.concurrent.FutureTask.run(FutureTask.java:266)     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)  at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)    ... 1 more ]
      
  </code></pre>