title: How we save a Huge pyspark dataframe?
body: <p>I have a big pyspark Dataframe which I want to save in <code>myfile</code> (.tsv)  for further use. To do that, I defined the following code:</p>
  <pre><code>with open(myfile, &quot;a&quot;) as csv_file:
          writer = csv.writer(csv_file, delimiter='\t')
          writer.writerow([&quot;vertex&quot; + &quot;\t&quot; + &quot;id_source&quot; + &quot;\t&quot; + &quot;id_target&quot; + &quot;\t&quot;+ &quot;similarity&quot;])
          
          for part_id in range(joinDesrdd_df.rdd.getNumPartitions()):
              part_rdd = joinDesrdd_df.rdd.mapPartitionsWithIndex(make_part_filter(part_id), True)
              data_from_part_rdd = part_rdd.collect()
              vertex_list = set()
             
              for row in data_from_part_rdd:
                  writer.writerow([....])
              
          csv_file.flush() 
  </code></pre>
  <p>My code cannot pass this step, It generates an exception:</p>
  <p>1.In the workers log:</p>
  <pre><code>19/07/22 08:58:57 INFO Worker: Executor app-20190722085320-0000/2 finished with state KILLED exitStatus 143
  14: 19/07/22 08:58:57 INFO ExternalShuffleBlockResolver: Application app-20190722085320-0000 removed, cleanupLocalDirs = true
  14: 19/07/22 08:58:57 INFO Worker: Cleaning up local directories for application app-20190722085320-0000
   5: 19/07/22 08:58:57 INFO Worker: Executor app-20190722085320-0000/1 finished with state KILLED exitStatus 143
   7: 19/07/22 08:58:57 INFO Worker: Executor app-20190722085320-0000/14 finished with state KILLED exitStatus 143
  ...
  </code></pre>
  <p>2.in the Job execution log:</p>
  <pre><code>Traceback (most recent call last):
    File &quot;/project/6008168/tamouze/RWLastVersion2207/module1.py&quot;, line 306, in &lt;module&gt;
      for part_id in range(joinDesrdd_df.rdd.getNumPartitions()):
    File &quot;/cvmfs/soft.computecanada.ca/easybuild/software/2017/Core/spark/2.3.0/python/lib/pyspark.zip/pyspark/sql/dataframe.py&quot;, line 88, in rdd
    File &quot;/cvmfs/soft.computecanada.ca/easybuild/software/2017/Core/spark/2.3.0/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py&quot;, line 1160, in __call__
    File &quot;/cvmfs/soft.computecanada.ca/easybuild/software/2017/Core/spark/2.3.0/python/lib/pyspark.zip/pyspark/sql/utils.py&quot;, line 63, in deco
    File &quot;/cvmfs/soft.computecanada.ca/easybuild/software/2017/Core/spark/2.3.0/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py&quot;, line 320, in get_return_value
  py4j.protocol.Py4JJavaError: An error occurred while calling o528.javaToPython.
  : org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree:
  Exchange hashpartitioning(id_source#263, id_target#292, similarity#258, 1024)
  +- *(11) HashAggregate(keys=[id_source#263, id_target#292, similarity#258], functions=[], output=[id_source#263, id_target#292, similarity#258])
  </code></pre>
  <p>I don't know why this piece of my code generates the exception. Note that on small data, the execution is ok, but with big data it's not.</p>
  <p>Also, please what is the best way to save a pysaprk dataframe for further use?</p>
  <p><strong>Update:</strong>
  I tried to replace the above with loop by the following:</p>
  <pre><code>joinDesrdd_df.withColumn(&quot;par_id&quot;,col('id_source')%50).repartition(50, 'par_id').write.format('parquet').partitionBy(&quot;par_id&quot;).save(&quot;/project/6008168/bib/RWLastVersion2207/randomWalkParquet/candidate.parquet&quot;)
  </code></pre>
  <p>I get similar exception:</p>
  <pre><code>19/07/22 21:10:18 INFO TaskSetManager: Finished task 653.0 in stage 11.0 (TID 2257) in 216940 ms on 172.16.140.237 (executor 14) (1017/1024)
  19/07/22 21:11:32 ERROR FileFormatWriter: Aborting job null.
  org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree:
  Exchange hashpartitioning(par_id#328, 50)
  +- *(12) HashAggregate(keys=[id_source#263, id_target#292, similarity#258], functions=[], output=[id_source#263, id_target#292, similarity#258, par_id#328])
     +- Exchange hashpartitioning(id_source#263, id_target#292, similarity#258, 1024)
        +- *(11) HashAggregate(keys=[id_source#263, id_target#292, similarity#258], functions=[], output=[id_source#263, id_target#292, similarity#258])
           +- *(11) Project [id_source#263, id_target#292, similarity#258]
              +- *(11) BroadcastHashJoin [instance_target#65], [instance#291], Inner, BuildRight
  </code></pre>