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>

<p>Update: It seems like your code was failing also due to the use of "count" as column name.
count seems to be protected keyword in DataFrame API.
renaming count to "mycount" solved the problem. The below working code was modify to support spark version 1.5.2 which I used to test your issue.</p>

<pre><code>df = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").load("/tmp/fac_cal.csv")
df1 = df.groupBy("name", "genre").count()
df1 = df1.select(col("name"),col("genre"),col("count").alias("mycount"))
df2 = df1.groupby('name').agg(F.max("mycount").alias("max_count"))
df2 = df2.select(col('name').alias('name2'),col("max_count"))
#Now trying to join both dataframes
final_df = df1.join(df2,[df1.name == df2.name2 , df1.mycount == df2.max_count])
final_df.show()

+-----+---------+-------+-----+---------+
| name|    genre|mycount|name2|max_count|
+-----+---------+-------+-----+---------+
|brata|   comedy|      2|brata|        2|
|brata|    drama|      2|brata|        2|
|panda|adventure|      1|panda|        1|
|panda|  romance|      1|panda|        1|
|satya|   action|      3|satya|        3|
+-----+---------+-------+-----+---------+
</code></pre>

<p>The example for complex condition in <a href="https://spark.apache.org/docs/2.0.0/api/python/pyspark.sql.html" rel="nofollow">https://spark.apache.org/docs/2.0.0/api/python/pyspark.sql.html</a></p>

<pre><code>cond = [df.name == df3.name, df.age == df3.age]
&gt;&gt;&gt; df.join(df3, cond, 'outer').select(df.name, df3.age).collect()
[Row(name=u'Alice', age=2), Row(name=u'Bob', age=5)]
</code></pre>

<hr>

<p>can you try: </p>

<pre><code>final_df = df1.join(df2, [df1.name == df2.name , df1.mycount == df2.max_count])
</code></pre>

<p>Note also, that according to the spec "left" is not part of the valid join types:
how â str, default âinnerâ. One of inner, outer, left_outer, right_outer, leftsemi.</p>

python apache-spark pyspark pyspark-sql