title: PySpark GC overhead limit exceeded and timeout
body: <p>I have written code in pyspark, task was to find Delta/changed records compared to previous day parquet file and write it to csv with additional column marking new records or existing record. This is done by concatenating the columns and applying base64(encode(columns here)) naming this as hash_key.</p>
  <ol>
  <li><p>Read 9 parquet files and registered temp tables against them.</p>
  <ul>
  <li>Apply a query over it, naming the output as df2.</li>
  <li>Read the previous day parquet file from S3 and name it as df1 and get count of it.</li>
  <li>Register temp tables against both dataframes as t1 and t2.</li>
  </ul>
  </li>
  <li><p>Now i used sqlContext.sql to find changed records from t2 compared to t1 using hash_key namely df3 and get the count of the rows.</p>
  </li>
  <li><p>Saving the different in csv file and replace the parquet file with data in df2.</p>
  </li>
  </ol>
  <p>Everything worked perfectly till here. But later the task was to show what rows are totally new in the delta records and what are old rows with some of the columns changed. Here i tried using PySpark code and Joined two dataframes df1 and df3. But i got following errors <strong>GC overhead limit exceeded</strong> or <strong>timeout</strong></p>
  <p>As an alternative i tried registering temp tables against the dataframes and executed sql query over it.
  but the issue remains the same. Am really stuck over it. Don't know why the resources are consumed.</p>
  <p>Configurations:</p>
  <pre><code>conf = (SparkConf()
          .setAppName(&quot;GD_Regex&quot;)
          .set(&quot;spark.executor.instances&quot;, &quot;3&quot;)
          .set(&quot;spark.executor.cores&quot;, &quot;3&quot;)
          .set(&quot;spark.sql.parquet.enableVectorizedReader&quot;, &quot;false&quot;)
          .set(&quot;spark.executor.memory&quot;, &quot;4g&quot;)
          .set(&quot;fs.s3a.server-side-encryption-algorithm&quot;, &quot;AES256&quot;)
          .set(&quot;spark.sql.autoBroadcastJoinThreshold&quot;,&quot;-1&quot;)
          )
  </code></pre>
  <p>Finding difference using hash_key</p>
  <pre><code>s3_parquet_file = sqlContext.read.parquet(parquet_path)
  historical_file_row_count = s3_parquet_file.count()
  s3_parquet_file.registerTempTable(&quot;s3_parquet_file_temp&quot;)
  query_response.registerTempTable(&quot;query_response_temp&quot;)
  filtered_data = sqlContext.sql(
          &quot;&quot;&quot;select * from query_response_temp where hash_key NOT IN ( SELECT hash_key FROM s3_parquet_file_temp )&quot;&quot;&quot;)
  </code></pre>
  <p>Approach 1:</p>
  <pre><code>ff = filtered_data.alias('df2').join(s3_parquet_file.alias('df1'), ( filtered_data.CTMS_STUDY_NUMBER == s3_parquet_file.CTMS_STUDY_NUMBER ) &amp; ( filtered_data.CTMS_SITE_NUMBER == s3_parquet_file.CTMS_SITE_NUMBER ) 
          &amp; (filtered_data.FULL_NAME_OF_PI == s3_parquet_file.FULL_NAME_OF_PI ) &amp; ( filtered_data.PRIMARY_ROLE_NAME == s3_parquet_file.PRIMARY_ROLE_NAME) &amp; ( filtered_data.CENTER_NAME == s3_parquet_file.CENTER_NAME ) , &quot;outer&quot; )\
          .select('df2.*', 'df1.CTMS_STUDY_NUMBER')\
          .withColumn('status', when(s3_parquet_file.CTMS_STUDY_NUMBER.isNotNull() &amp; filtered_data.CTMS_STUDY_NUMBER.isNotNull(), 'existing').otherwise('new'))\
          .filter(filtered_data.CTMS_STUDY_NUMBER.isNotNull())\
          .select('df2.*', 'status')
  </code></pre>
  <p>Approach 2:</p>
  <pre><code>q = &quot;&quot;&quot;SELECT query_response_temp.*, IF( query_response_temp.CTMS_STUDY_NUMBER IS NOT NULL AND s3_parquet_file_temp.CTMS_STUDY_NUMBER IS NOT NULL, &quot;existing&quot;, &quot;new&quot; ) as status FROM query_response_temp FULL OUTER JOIN s3_parquet_file_temp on query_response_temp.CTMS_STUDY_NUMBER = s3_parquet_file_temp.CTMS_STUDY_NUMBER AND query_response_temp.CTMS_SITE_NUMBER = s3_parquet_file_temp.CTMS_SITE_NUMBER AND query_response_temp.FULL_NAME_OF_PI = s3_parquet_file_temp.FULL_NAME_OF_PI AND query_response_temp.PRIMARY_ROLE_NAME = s3_parquet_file_temp.PRIMARY_ROLE_NAME AND query_response_temp.CENTER_NAME = s3_parquet_file_temp.CENTER_NAME&quot;&quot;&quot;
          delta_records = sqlContext.sql(q)
          delta_records = delta_records.filter(delta_records.CTMS_STUDY_NUMBER.isNotNull())
  </code></pre>
  <p>Code where the error arrives when it hits action:</p>
  <pre><code>dataset = dataset.drop(&quot;hash_key&quot;)
  print('hash_key dropped.')
  dataset.coalesce(1).write.format('csv') \
    .option('header', 'true') \
    .option(&quot;compression&quot;, &quot;none&quot;) \
    .mode('overwrite') \
    .save(_path)
  </code></pre>