python+pyspark: error on inner join with multiple column comparison in pyspark
<p>Hi I have 2 dataframes to join</p>

<pre><code>#df1
 name    genre  count
 satya   drama    1
 satya   action   3
 abc     drame    2
 abc     comedy   2
 def     romance  1

#df2
 name  max_count
 satya  3
 abc    2
 def    1
</code></pre>

<p>Now I want to join above 2 dfs on name and count==max_count, But i am getting an error</p>

<pre><code>import pyspark.sql.functions as F
from pyspark.sql.functions import count, col
from pyspark.sql.functions import struct
df = spark.read.csv('file',sep = '###', header=True)
df1 = df.groupBy("name", "genre").count()
df2 = df1.groupby('name').agg(F.max("count").alias("max_count"))
#Now trying to join both dataframes
final_df = df1.join(df2, (df1.name == df2.name) &amp; (df1.count == df2.max_count))
final_df.show() ###Error
#py4j.protocol.Py4JJavaError: An error occurred while calling o207.showString.
: org.apache.spark.SparkException: Exception thrown in awaitResult:
    at org.apache.spark.util.ThreadUtils$.awaitResult(ThreadUtils.scala:194)
#Caused by: java.lang.UnsupportedOperationException: Cannot evaluate expression: count(1)
    at org.apache.spark.sql.catalyst.expressions.Unevaluable$class.doGenCode(Expression.scala:224)
</code></pre>

<p>But success with "left " join</p>

<pre><code>final_df = df1.join(df2, (df1.name == df2.name) &amp; (df1.count == df2.max_count), "left")
final_df.show()  ###Success but i don't want left join , i want inner join
</code></pre>

<p>My question is why the above one fails, am I doing something wrong there???</p>

<p>I referred this link "<a href="http://stackoverflow.com/questions/35218882/find-maximum-row-per-group-in-spark-dataframe">Find maximum row per group in Spark DataFrame</a>". Used the first answer (2 groupby method).But same error.</p>

<p>I am on spark-2.0.0-bin-hadoop2.7 and python 2.7.</p>

<p>Please suggest.Thanks.</p>

<h1>Edit:</h1>

<p>The above scenario works with spark 1.6 (which is quite surprising that what's wrong with spark 2.0 (or with my installation , I will reinstall, check and update here)).</p>

<p>Has anybody tried this on spark 2.0 and got success , by following Yaron's answer below???</p>

<h1>My work-around in spark 2.0</h1>

<p>I created a single column('combined') from columns in join comparision('name','mycount')in respective dfs, so now I have one column to compare and this is not creating any issue as I am comparing only one column.</p>

<pre><code>def combine_func(*args):
  data = '_'.join([str(x) for x in args]) ###converting nonstring to str tehn concatenation
  return data
combine_func = udf(combine_func, StringType())  ##register the func as udf
df1 = df1.withColumn('combined_new_1', combine_new(df1['name'],df1['mycount']))  ###a col having concatenated value from name and mycount columns eg: 'satya_3'
df2 = df2.withColumn('combined_new_2', combine_new(df2['name2'],df2['max_count']))
#df1.columns == 'name','genre', 'mycount', 'combined_new_1'
#df2.columns == 'name2', 'max_count', 'combined_new_2'
#Now join 
final_df = df1.join(df2,df1.combined_new_1 == df2.combined_new_2, 'inner')
#final_df = df1.join(df2,df1.combined_new_1 == df2.combined_new_2, 'inner').select('the columns you want')
final_df.show()  ####It is showing the result, Trust me.
</code></pre>

<p>Please don't follow until unless you are in a hurry, Better search for a reliable solution.</p>

python apache-spark pyspark pyspark-sql