title: PySpark error: Py4JJavaError: An error occurred while calling o469.count
body: <p>I have a python spark program which behaves somewhere inconsistently and in some cases errors.<br>
  I usually run in on a small EMR cluster of two <code>c3.2xlarge</code> with a <code>m1.large</code> master and it runs fine and competes successfully.<br>
  However, when I run the exact same program on a larger cluster - I tried 4 <code>c3.2xlarge</code> with a <code>m1.large</code> master, it ends with an error.  I'll paste the error(s) below, but even these errors aren't quite consistent, not the error trace itself and not the stage at which it happens.<br>
  In one case for example, it happens after about 26min and on a <code>.count()</code> call and in and in another instance it actually passes the <code>.count()</code> successfully, but an error happens after about an hour and in a different stage, on a call to <code>.write.jdbc()</code>.<br>
  So I assume this is a kind of race condition but I'm not even sure whether it's my fault for improper usage of spark, or is this a bug in spark.<br>
  Most of the features I use in this case are from spark.sql.</p>
  
  <p>Environment: Spark 1.5.2 on EMR (Elastic Mapreduce on AWS)</p>
  
  <p>The stack traces are very long, so I can't paste the entire traces here, but hopefully just enough to get the context.<br>
  The code itself - well, there's a lot of it and I hadn't managed to find a simple repro test case that I can easily post here... (race conditions, you know...)</p>
  
  <p>As noted, that's just part of the stack trace, it gets pretty long (example one):<br>
  Note that the error in both cases happens in different locations of the code. </p>
  
  <p><strong>Any help or directions how to approach this problem?</strong> </p>
  
  <p>cheers</p>
  
  <pre><code>Traceback (most recent call last):
    File "/home/hadoop/rantav.spark_normalize_data.py.134231/spark_normalize_data.py", line 102, in &lt;module&gt;
      run_spark(args)
    File "/home/hadoop/rantav.spark_normalize_data.py.134231/spark_normalize_data.py", line 62, in run_spark
      company_aliases_broadcast, experiences, args)
    File "/home/hadoop/rantav.spark_normalize_data.py.134231/companies.py", line 50, in get_companies
      out(sc, args, 'companies', sql_create_table, companies)
    File "/home/hadoop/rantav.spark_normalize_data.py.134231/output.py", line 48, in out
      mode='append')
    File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/readwriter.py", line 455, in jdbc
    File "/usr/lib/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 538, in __call__
    File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 36, in deco
    File "/usr/lib/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py", line 300, in get_return_value
  py4j.protocol.Py4JJavaError: An error occurred while calling o464.jdbc.
  : org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree:
  TungstenExchange hashpartitioning(ref_company_size_id#96)
   TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,churn_rate#200,churn_rate_percentile#202,retention_rate_2y#210]
    SortMergeOuterJoin [id#85], [company_id#180], LeftOuter, None
     TungstenSort [id#85 ASC], false, 0
      TungstenExchange hashpartitioning(id#85)
       TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,churn_rate#200,(_we0#203 * 100.0) AS churn_rate_percentile#202]
        Window [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,churn_rate#200], [HiveWindowFunction#org.apache.hadoop.hive.ql.udf.generic.GenericUDAFPercentRank(churn_rate#200) WindowSpecDefinition ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS _we0#203], [ref_company_size_id#96], [churn_rate#200 ASC]
         TungstenSort [ref_company_size_id#96 ASC,churn_rate#200 ASC], false, 0
          TungstenExchange hashpartitioning(ref_company_size_id#96)
           TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,(100.0 * cast(pythonUDF#201 as double)) AS churn_rate#200]
            !BatchPythonEvaluation PythonUDF#divide(count#199L,emp_count#94L), [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,count#199L,pythonUDF#201]
             ConvertToSafe
              TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,count#199L]
               SortMergeOuterJoin [id#85], [company_id#180], LeftOuter, None
                TungstenSort [id#85 ASC], false, 0
                 TungstenExchange hashpartitioning(id#85)
                  TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,(_we0#198 * 100.0) AS avg_tenure_percentile#197]
                   Window [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196], [HiveWindowFunction#org.apache.hadoop.hive.ql.udf.generic.GenericUDAFPercentRank(avg_tenure#196) WindowSpecDefinition ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS _we0#198], [ref_company_size_id#96], [avg_tenure#196 ASC]
                    TungstenSort [ref_company_size_id#96 ASC,avg_tenure#196 ASC], false, 0
                     TungstenExchange hashpartitioning(ref_company_size_id#96)
                      TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg(duration_days)#195 AS avg_tenure#196]
                       SortMergeOuterJoin [id#85], [company_id#180], LeftOuter, None
                        TungstenSort [id#85 ASC], false, 0
                         TungstenExchange hashpartitioning(id#85)
                          TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,(_we0#194 * 100.0) AS growth_rate_percentile#193]
                           Window [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191], [HiveWindowFunction#org.apache.hadoop.hive.ql.udf.generic.GenericUDAFPercentRank(growth_rate#191) WindowSpecDefinition ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS _we0#194], [ref_company_size_id#96], [growth_rate#191 ASC]
                            TungstenSort [ref_company_size_id#96 ASC,growth_rate#191 ASC], false, 0
                             TungstenExchange hashpartitioning(ref_company_size_id#96)
                              TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,(100.0 * cast(pythonUDF#192 as double)) AS growth_rate#191]
                               !BatchPythonEvaluation PythonUDF#divide(count#190L,emp_count#94L), [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,count#190L,pythonUDF#192]
                                ConvertToSafe
                                 TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,count#190L]
                                  SortMergeOuterJoin [id#85], [company_id#180], LeftOuter, None
                                   TungstenSort [id#85 ASC], false, 0
                                    TungstenExchange hashpartitioning(id#85)
                                     TungstenProject [id#85,href#86,name#87,emp_count#94L,(_we0#176 * 100.0) AS emp_count_percentile#175,ref_company_size_id#96]
                                      Window [id#85,href#86,name#87,emp_count#94L,ref_company_size_id#96], [HiveWindowFunction#org.apache.hadoop.hive.ql.udf.generic.GenericUDAFPercentRank(emp_count#94L) WindowSpecDefinition ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS _we0#176], [ref_company_size_id#96], [emp_count#94L ASC]
                                       TungstenSort [ref_company_size_id#96 ASC,emp_count#94L ASC], false, 0
  </code></pre>
  
  <p>And here's another example of a different stack trace (same program, same number of workers):</p>
  
  <pre><code>  /home/hadoop/rantav.spark_normalize_data.py.093920/pymysql/cursors.py:146: Warning: Can't create database 'v2'; database exists
      result = self._query(query)
    /home/hadoop/rantav.spark_normalize_data.py.093920/pymysql/cursors.py:146: Warning: Table 'oplog' already exists
      result = self._query(query)
    Traceback (most recent call last):
      File "/home/hadoop/rantav.spark_normalize_data.py.093920/spark_normalize_data.py", line 102, in &lt;module&gt;
        run_spark(args)
      File "/home/hadoop/rantav.spark_normalize_data.py.093920/spark_normalize_data.py", line 62, in run_spark
        company_aliases_broadcast, experiences, args)
      File "/home/hadoop/rantav.spark_normalize_data.py.093920/companies.py", line 50, in get_companies
        out(sc, args, 'companies', sql_create_table, companies)
      File "/home/hadoop/rantav.spark_normalize_data.py.093920/output.py", line 35, in out
        if data.count() &gt; 0:
      File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py", line 268, in count
      File "/usr/lib/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 538, in __call__
      File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 36, in deco
      File "/usr/lib/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py", line 300, in get_return_value
    py4j.protocol.Py4JJavaError: An error occurred while calling o469.count.
    : org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree:
    TungstenAggregate(key=[], functions=[(count(1),mode=Final,isDistinct=false)], output=[count#216L])
     TungstenExchange SinglePartition
      TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=[currentCount#219L])
       TungstenProject
        Window [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,churn_rate#200,churn_rate_percentile#202,retention_rate_2y#210], [HiveWindowFunction#org.apache.hadoop.hive.ql.udf.generic.GenericUDAFPercentRank(retention_rate_2y#210) WindowSpecDefinition ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS _we0#213], [ref_company_size_id#96], [retention_rate_2y#210 ASC]
         TungstenSort [ref_company_size_id#96 ASC,retention_rate_2y#210 ASC], false, 0
          TungstenExchange hashpartitioning(ref_company_size_id#96)
           TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,churn_rate#200,churn_rate_percentile#202,retention_rate_2y#210]
            SortMergeOuterJoin [id#85], [company_id#180], LeftOuter, None
             TungstenSort [id#85 ASC], false, 0
              TungstenExchange hashpartitioning(id#85)
               TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,churn_rate#200,(_we0#203 * 100.0) AS churn_rate_percentile#202]
                Window [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,churn_rate#200], [HiveWindowFunction#org.apache.hadoop.hive.ql.udf.generic.GenericUDAFPercentRank(churn_rate#200) WindowSpecDefinition ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS _we0#203], [ref_company_size_id#96], [churn_rate#200 ASC]
                 TungstenSort [ref_company_size_id#96 ASC,churn_rate#200 ASC], false, 0
                  TungstenExchange hashpartitioning(ref_company_size_id#96)
                   TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,(100.0 * cast(pythonUDF#201 as double)) AS churn_rate#200]
                    !BatchPythonEvaluation PythonUDF#divide(count#199L,emp_count#94L), [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,count#199L,pythonUDF#201]
                     ConvertToSafe
                      TungstenProject [id#85,href#86,name#87,emp_count#94L,emp_count_percentile#175,ref_company_size_id#96,growth_rate#191,growth_rate_percentile#193,avg_tenure#196,avg_tenure_percentile#197,count#199L]
                       SortMergeOuterJoin [id#85], [company_id#180], LeftOuter, None
                        TungstenSort [id#85 ASC], false, 0
                         TungstenExchange hashpartitioning(id#85)
  </code></pre>