title: Spark distinct performance issue after joining
body: <p>I  reading a parquet file</p>
  
  <pre><code>val df1 = sqlContext.read.parquet("")
  df1.count = 5000000 records
  val df2 = df1.select("id","aid", "DId" , "dd" , "mm", "yy","TO")
    .distinct()
    .groupBy("id","aid", "DId" , "dd" , "mm","yy","TO")
    .count()
    .filter($"count" &gt; 1)
    .select("id","aid", "DId" , "dd" , "mm", "yy")
  val df3 = df2.join(df1,
    df2("cid") &lt;=&gt; df1("cid") &amp;&amp; 
    df2("aid") &lt;=&gt; df1("aid") &amp;&amp; 
    df2("did") &lt;=&gt; df1("did") &amp;&amp; 
    df2("dd") &lt;=&gt; df1("dd") &amp;&amp; 
    df2("mm") &lt;=&gt; df1("mm") &amp;&amp; 
    df2("yy") &lt;=&gt; df1("yy"), "inner")
    .distinct()
  </code></pre>
  
  <p>when I do df3.count ... It is taking nearly 897517ms which is too high it is impacting overall job execution and some times job is aborted If I Increase the timeout option it is execution  but latency is too high . I need suggestion to improve this</p>