title: Very long spark process on EMR
body: <p><strong>Context</strong>
  I am processing some data (5 billion rows, ~7 columns) via pyspark on EMR.</p>
  <p>The first steps including some joins work as expected, up to and including a <code>cache()</code> (memory_disk_ser). Then I filter one column for nulls and do a count() of this big dataframe.</p>
  <p><strong>Problem</strong></p>
  <p>It takes hours to then fail with a 'no connection error' (I do not remember precisely, but I am more interested in the 'why' of it being slow than the final error).</p>
  <p><strong>What I noticed</strong></p>
  <p>Out of my 256 vcores, 1 is always at 100%, the rest is idle. The one at 100% is used by a data node JVM.</p>
  <p><strong>Configuration</strong></p>
  <p>I have 4 r5a.16xlarge instances, each with 4 EBS ssds.</p>
  <p>EMR is supposed to take care of its own config, and that is what I see in the spark UI:</p>
  <ul>
  <li>spark.emr.default.executor.memory 18971M</li>
  <li>spark.driver.memory   2048M</li>
  <li>spark.executor.cores  4</li>
  </ul>
  <p>I am setting myself:</p>
  <ul>
  <li>spark.network.timeout: 800s</li>
  <li>spark.executor.heartbeatInterval: 60s</li>
  <li>spark.dynamicAllocation.enabled: True</li>
  <li>spark.dynamicAllocation.shuffleTracking.enabled: True</li>
  <li>spark.executor.instances: 0</li>
  <li>spark.default.parallelism: 128</li>
  <li>spark.shuffle.spill.compress: true</li>
  <li>spark.shuffle.compress: true</li>
  <li>spark.rdd.compress: true</li>
  <li>spark.storage.level: MEMORY_AND_DISK_SER</li>
  <li>spark.executor.extraJavaOptions: -X:+seG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' -Duser.timezone=GMT</li>
  <li>spark.driver.extraJavaOptions: -XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' -Duser.timezone=GMT</li>
  </ul>
  <p><strong>Question</strong></p>
  <p>What do I do wrong, or what do I not understand properly? Counting a cached dataframe built in 10 minutes, even when filtering out nulls, should not take hours.</p>
  <p><strong>Some more details</strong>
  The data source is on S3, on homogeneous parquet files. But reading those always work fine, because the join succeeds.
  During the count(), I see 200 taaks, 195 succeeds within a few seconds, but 5 consistently never complete, all NODE_LOCAL (but some NODE_LOCAL) tasks do complete</p>