title: Pyspark - distinct records based on 2 columns in dataframe
body: <p>I have 2 dataframes, say <code>df1</code> and <code>df2</code>.</p>
  
  <p><code>df1</code> data comes from a database, and <code>df2</code> is the new data I receive from my client.  I need to process the new data, and perform <code>UPSERTs</code> based on whether it is a new record or an existing record to be updated.</p>
  
  <p>Sample data output:</p>
  
  <pre><code>df1= sqlContext.createDataFrame([("xxx1","81A01","TERR NAME 01","NJ"),("xxx2","81A01","TERR NAME 01","NJ"),("xxx3","81A01","TERR NAME 01","NJ"),("xxx4","81A01","TERR NAME 01","CA"),("xx5","81A01","TERR NAME 01","ME")], ["zip_code","territory_code","territory_name","state"])
  df2= sqlContext.createDataFrame([("xxx1","81A01","TERR NAME 55","NY"),("xxx2","81A01","TERR NAME 55","NY"),("x103","81A01","TERR NAME 01","NJ")], ["zip_code","territory_code","territory_name","state"])
  
  df1.show()
  +--------+--------------+--------------+-----+
  |zip_code|territory_code|territory_name|state|
  +--------+--------------+--------------+-----+
  |    xxx1|         81A01|  TERR NAME 01|   NJ|
  |    xxx2|         81A01|  TERR NAME 01|   NJ|
  |    xxx3|         81A01|  TERR NAME 01|   NJ|
  |    xxx4|         81A01|  TERR NAME 01|   CA|
  |    xxx5|         81A01|  TERR NAME 01|   ME|
  +---------------------------------------------
  
  # Print out information about this data
  df2.show()
  +--------+--------------+--------------+-----+
  |zip_code|territory_code|territory_name|state|
  +--------+--------------+--------------+-----+
  |    xxx1|         81A01|  TERR NAME 55|   NY|
  |    xxx2|         81A01|  TERR NAME 55|   NY|
  |    x103|         81A01|  TERR NAME 01|   NJ|
  +---------------------------------------------
  </code></pre>
  
  <p>Expected results:
  I need to compare <strong>df2</strong> dataframe with <strong>df1</strong>.
  Create 2 new datasets based on above compare, i.e., records to be updated &amp; records to be appended / inserted  to database.</p>
  
  <p>if <strong>zip_code</strong> &amp; <strong>territory_code</strong> are same, then it is an UPDATE, else it is an INSERT to database.</p>
  
  <p>For example:
  New dataframe output for INSERT:</p>
  
  <pre><code> +--------+--------------+--------------+-----+
   |zip_code|territory_code|territory_name|state|
   +--------+--------------+--------------+-----+
   |    x103|         81A01|  TERR NAME 01|   NJ|
   +---------------------------------------------
  </code></pre>
  
  <p>New dataframe for UPDATE:</p>
  
  <pre><code>+--------+--------------+--------------+-----+
  |zip_code|territory_code|territory_name|state|
  +--------+--------------+--------------+-----+
  |    xxx1|         81A01|  TERR NAME 55|   NY|
  |    xxx2|         81A01|  TERR NAME 55|   NY|
  +---------------------------------------------
  </code></pre>
  
  <p>Can someone please help me? And I am using AWS Glue.</p>
  
  <p>UPDATE: SOLUTION (Using join &amp; subtract)</p>
  
  <pre><code>df3 = df1.join(df2, (df1.zip_code == df2.zip_code_new) &amp; (df1.territory_code == df2.territory_code_new))
  df5=df3.drop("zip_code", "territory_code", "territory_name", "state")
  df5.show()
  
  +------------+------------------+------------------+---------+
  |zip_code_new|territory_code_new|territory_name_new|state_new|
  +------------+------------------+------------------+---------+
  |        x103|             81A01|      TERR NAME 01|       NJ|
  +------------+------------------+------------------+---------+
  
  df4=df2.subtract(df5)
  df4.show()
  
  +------------+------------------+------------------+---------+
  |zip_code_new|territory_code_new|territory_name_new|state_new|
  +------------+------------------+------------------+---------+
  |    xxx1    |         81A01    |  TERR NAME 55    |   NY    |
  |    xxx2    |         81A01    |  TERR NAME 55    |   NY    |
  +------------------------------------------------------------+
  </code></pre>
  
  <p>For RDS database updation, I use pymysql/Mysqldb:</p>
  
  <pre><code>db = MySQLdb.connect("xxxx.rds.amazonaws.com", "username", "password", "databasename")
  cursor = db.cursor()
  
  #cursor.execute("REPLACE INTO table SELECT * FROM table_stg")
  insertQry = "INSERT INTO table VALUES('xxx1','81A01','TERR NAME 55','NY') ON DUPLICATE KEY UPDATE territory_name='TERR NAME 55', state='NY'"
  n=cursor.execute(insertQry)
  db.commit()
  cursor.fetchall()
  db.close()
  </code></pre>
  
  <p>Thanks</p>